# Source code for opentelemetry.sdk.metrics._internal.aggregation

    # Copyright The OpenTelemetry Authors
    # SPDX-License-Identifier: Apache-2.0

    # pylint: disable=too-many-lines

    import math
    from abc import ABC, abstractmethod
    from bisect import bisect_left
    from collections.abc import Callable, Sequence
    from enum import IntEnum
    from functools import partial
    from logging import getLogger
    from threading import Lock
    from typing import (
        Generic,
        TypeVar,
        cast,
    )

    from opentelemetry.metrics import (
        Asynchronous,
        Counter,
        Histogram,
        ObservableCounter,
        ObservableGauge,
        ObservableUpDownCounter,
        Synchronous,
        UpDownCounter,
        _Gauge,
    )
    from opentelemetry.sdk.metrics._internal.exemplar import (
        Exemplar,
        ExemplarReservoirBuilder,
    )
    from opentelemetry.sdk.metrics._internal.exponential_histogram.buckets import (
        Buckets,
    )
    from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping import (
        Mapping,
    )
    from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping.exponent_mapping import (
        ExponentMapping,
    )
    from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping.logarithm_mapping import (
        LogarithmMapping,
    )
    from opentelemetry.sdk.metrics._internal.instrument import _Instrument
    from opentelemetry.sdk.metrics._internal.measurement import Measurement
    from opentelemetry.sdk.metrics._internal.point import Buckets as BucketsPoint
    from opentelemetry.sdk.metrics._internal.point import (
        ExponentialHistogramDataPoint,
        HistogramDataPoint,
        NumberDataPoint,
        Sum,
    )
    from opentelemetry.sdk.metrics._internal.point import Gauge as GaugePoint
    from opentelemetry.sdk.metrics._internal.point import (
        Histogram as HistogramPoint,
    )
    from opentelemetry.util.types import Attributes

    _DataPointVarT = TypeVar("_DataPointVarT", NumberDataPoint, HistogramDataPoint)

    _logger = getLogger(__name__)

    class AggregationTemporality(IntEnum):
        """
        The temporality to use when aggregating data.

        Can be one of the following values:
        """

        UNSPECIFIED = 0
        DELTA = 1
        CUMULATIVE = 2

    class _Aggregation(ABC, Generic[_DataPointVarT]):
        def __init__(
            self,
            attributes: Attributes,
            reservoir_builder: ExemplarReservoirBuilder,
        ):
            self._lock = Lock()
            self._attributes = attributes
            self._reservoir = reservoir_builder()
            self._previous_point = None

        @abstractmethod
        def aggregate(
            self, measurement: Measurement, should_sample_exemplar: bool = True
        ) -> None:
            """Aggregate a measurement.

            Args:
                measurement: Measurement to aggregate
                should_sample_exemplar: Whether the measurement should be sampled by the exemplars reservoir or not.
            """

        @abstractmethod
        def collect(
            self,
            collection_aggregation_temporality: AggregationTemporality,
            collection_start_nano: int,
        ) -> _DataPointVarT | None:
            pass

        def _collect_exemplars(self) -> Sequence[Exemplar]:
            """Returns the collected exemplars.

            Returns:
                The exemplars collected by the reservoir
            """
            return self._reservoir.collect(self._attributes)

        def _sample_exemplar(
            self, measurement: Measurement, should_sample_exemplar: bool
        ) -> None:
            """Offer the measurement to the exemplar reservoir for sampling.

            It should be called within the each :ref:`aggregate` call.

            Args:
                measurement: The new measurement
                should_sample_exemplar: Whether the measurement should be sampled by the exemplars reservoir or not.
            """
            if should_sample_exemplar:
                self._reservoir.offer(
                    measurement.value,
                    measurement.time_unix_nano,
                    measurement.attributes,
                    measurement.context,
                )

    class _DropAggregation(_Aggregation):
        def aggregate(
            self, measurement: Measurement, should_sample_exemplar: bool = True
        ) -> None:
            pass

        def collect(
            self,
            collection_aggregation_temporality: AggregationTemporality,
            collection_start_nano: int,
        ) -> _DataPointVarT | None:
            pass

    class _SumAggregation(_Aggregation[Sum]):
        def __init__(
            self,
            attributes: Attributes,
            instrument_is_monotonic: bool,
            instrument_aggregation_temporality: AggregationTemporality,
            start_time_unix_nano: int,
            reservoir_builder: ExemplarReservoirBuilder,
        ):
            super().__init__(attributes, reservoir_builder)

            self._start_time_unix_nano = start_time_unix_nano
            self._instrument_aggregation_temporality = (
                instrument_aggregation_temporality
            )
            self._instrument_is_monotonic = instrument_is_monotonic

            self._value = None

            self._previous_collection_start_nano = self._start_time_unix_nano
            self._previous_value = 0

        def aggregate(
            self, measurement: Measurement, should_sample_exemplar: bool = True
        ) -> None:
            with self._lock:
                if self._value is None:
                    self._value = 0

                self._value = self._value + measurement.value

            self._sample_exemplar(measurement, should_sample_exemplar)

        def collect(
            self,
            collection_aggregation_temporality: AggregationTemporality,
            collection_start_nano: int,
        ) -> NumberDataPoint | None:
            """
            Atomically return a point for the current value of the metric and
            reset the aggregation value.

            Synchronous instruments have a method which is called directly with
            increments for a given quantity:

            For example, an instrument that counts the amount of passengers in
            every vehicle that crosses a certain point in a highway:

            synchronous_instrument.add(2)
            collect(...)  # 2 passengers are counted
            synchronous_instrument.add(3)
            collect(...)  # 3 passengers are counted
            synchronous_instrument.add(1)
            collect(...)  # 1 passenger is counted

            In this case the instrument aggregation temporality is DELTA because
            every value represents an increment to the count,

            Asynchronous instruments have a callback which returns the total value
            of a given quantity:

            For example, an instrument that measures the amount of bytes written to
            a certain hard drive:

            callback() -> 1352
            collect(...) # 1352 bytes have been written so far
            callback() -> 2324
            collect(...) # 2324 bytes have been written so far
            callback() -> 4542
            collect(...) # 4542 bytes have been written so far

            In this case the instrument aggregation temporality is CUMULATIVE
            because every value represents the total of the measurement.

            There is also the collection aggregation temporality, which is passed
            to this method. The collection aggregation temporality defines the
            nature of the returned value by this aggregation.

            When the collection aggregation temporality matches the
            instrument aggregation temporality, then this method returns the
            current value directly:

            synchronous_instrument.add(2)
            collect(DELTA) -> 2
            synchronous_instrument.add(3)
            collect(DELTA) -> 3
            synchronous_instrument.add(1)
            collect(DELTA) -> 1

            callback() -> 1352
            collect(CUMULATIVE) -> 1352
            callback() -> 2324
            collect(CUMULATIVE) -> 2324
            callback() -> 4542
            collect(CUMULATIVE) -> 4542

            When the collection aggregation temporality does not match the
            instrument aggregation temporality, then a conversion is made. For this
            purpose, this aggregation keeps a private attribute,
            self._previous_value.

            When the instrument is synchronous:

            self._previous_value is the sum of every previously
            collected (delta) value. In this case, the returned (cumulative) value
            will be:

            self._previous_value + value

            synchronous_instrument.add(2)
            collect(CUMULATIVE) -> 2
            synchronous_instrument.add(3)
            collect(CUMULATIVE) -> 5
            synchronous_instrument.add(1)
            collect(CUMULATIVE) -> 6

            Also, as a diagram:

            time ->

            self._previous_value
            |-------------|

            value (delta)
                          |----|

            returned value (cumulative)
            |------------------|

            When the instrument is asynchronous:

            self._previous_value is the value of the previously
            collected (cumulative) value. In this case, the returned (delta) value
            will be:

            value - self._previous_value

            callback() -> 1352
            collect(DELTA) -> 1352
            callback() -> 2324
            collect(DELTA) -> 972
            callback() -> 4542
            collect(DELTA) -> 2218

            Also, as a diagram:

            time ->

            self._previous_value
            |-------------|

            value (cumulative)
            |------------------|

            returned value (delta)
                          |----|
            """

            with self._lock:
                value = self._value
                self._value = None

                if (
                    self._instrument_aggregation_temporality
                    is AggregationTemporality.DELTA
                ):
                    # This happens when the corresponding instrument for this
                    # aggregation is synchronous.
                    if (
                        collection_aggregation_temporality
                        is AggregationTemporality.DELTA
                    ):
                        previous_collection_start_nano = (
                            self._previous_collection_start_nano
                        )
                        self._previous_collection_start_nano = (
                            collection_start_nano
                        )

                        if value is None:
                            return None

                        return NumberDataPoint(
                            attributes=self._attributes,
                            exemplars=self._collect_exemplars(),
                            start_time_unix_nano=previous_collection_start_nano,
                            time_unix_nano=collection_start_nano,
                            value=value,
                        )

                    if value is None:
                        value = 0

                    self._previous_value = value + self._previous_value

                    return NumberDataPoint(
                        attributes=self._attributes,
                        exemplars=self._collect_exemplars(),
                        start_time_unix_nano=self._start_time_unix_nano,
                        time_unix_nano=collection_start_nano,
                        value=self._previous_value,
                    )

                # This happens when the corresponding instrument for this
                # aggregation is asynchronous.

                if value is None:
                    # This happens when the corresponding instrument callback
                    # does not produce measurements.
                    return None

                if (
                    collection_aggregation_temporality
                    is AggregationTemporality.DELTA
                ):
                    result_value = value - self._previous_value

                    self._previous_value = value

                    previous_collection_start_nano = (
                        self._previous_collection_start_nano
                    )
                    self._previous_collection_start_nano = collection_start_nano

                    return NumberDataPoint(
                        attributes=self._attributes,
                        exemplars=self._collect_exemplars(),
                        start_time_unix_nano=previous_collection_start_nano,
                        time_unix_nano=collection_start_nano,
                        value=result_value,
                    )

                return NumberDataPoint(
                    attributes=self._attributes,
                    exemplars=self._collect_exemplars(),
                    start_time_unix_nano=self._start_time_unix_nano,
                    time_unix_nano=collection_start_nano,
                    value=value,
                )

    class _LastValueAggregation(_Aggregation[GaugePoint]):
        def __init__(
            self,
            attributes: Attributes,
            reservoir_builder: ExemplarReservoirBuilder,
        ):
            super().__init__(attributes, reservoir_builder)
            self._value = None

        def aggregate(
            self, measurement: Measurement, should_sample_exemplar: bool = True
        ):
            with self._lock:
                self._value = measurement.value

            self._sample_exemplar(measurement, should_sample_exemplar)

        def collect(
            self,
            collection_aggregation_temporality: AggregationTemporality,
            collection_start_nano: int,
        ) -> _DataPointVarT | None:
            """
            Atomically return a point for the current value of the metric.
            """
            with self._lock:
                if self._value is None:
                    return None
                value = self._value
                self._value = None

            exemplars = self._collect_exemplars()

            return NumberDataPoint(
                attributes=self._attributes,
                exemplars=exemplars,
                start_time_unix_nano=None,
                time_unix_nano=collection_start_nano,
                value=value,
            )

    _DEFAULT_EXPLICIT_BUCKET_HISTOGRAM_AGGREGATION_BOUNDARIES: Sequence[float] = (
        0.0,
        5.0,
        10.0,
        25.0,
        50.0,
        75.0,
        100.0,
        250.0,
        500.0,
        750.0,
        1000.0,
        2500.0,
        5000.0,
        7500.0,
        10000.0,
    )

    class _ExplicitBucketHistogramAggregation(_Aggregation[HistogramPoint]):
        def __init__(
            self,
            attributes: Attributes,
            instrument_aggregation_temporality: AggregationTemporality,
            start_time_unix_nano: int,
            reservoir_builder: ExemplarReservoirBuilder,
            boundaries: Sequence[float] | None = None,
            record_min_max: bool = True,
        ):
            if boundaries is None:
                boundaries = (
                    _DEFAULT_EXPLICIT_BUCKET_HISTOGRAM_AGGREGATION_BOUNDARIES
                )
            if boundaries:
                for idx, bound in enumerate(boundaries):
                    if not math.isfinite(bound):
                        raise ValueError(f"invalid boundary: {bound!r}")
                    if idx and boundaries[idx - 1] >= bound:
                        raise ValueError("boundaries must be strictly increasing")
            super().__init__(
                attributes,
                reservoir_builder=partial(
                    reservoir_builder, boundaries=boundaries
                ),
            )

            self._instrument_aggregation_temporality = (
                instrument_aggregation_temporality
            )
            self._start_time_unix_nano = start_time_unix_nano
            self._boundaries = tuple(boundaries)
            self._record_min_max = record_min_max

            self._value = None
            self._min = math.inf
            self._max = -math.inf
            self._sum = 0

            self._previous_value = None
            self._previous_min = math.inf
            self._previous_max = -math.inf
            self._previous_sum = 0

            self._previous_collection_start_nano = self._start_time_unix_nano

        def _get_empty_bucket_counts(self) -> list[int]:
            return [0] * (len(self._boundaries) + 1)

        def aggregate(
            self, measurement: Measurement, should_sample_exemplar: bool = True
        ) -> None:
            with self._lock:
                if self._value is None:
                    self._value = self._get_empty_bucket_counts()

                measurement_value = measurement.value

                self._sum += measurement_value

                if self._record_min_max:
                    self._min = min(self._min, measurement_value)
                    self._max = max(self._max, measurement_value)

                self._value[bisect_left(self._boundaries, measurement_value)] += 1

            self._sample_exemplar(measurement, should_sample_exemplar)

        def collect(
            self,
            collection_aggregation_temporality: AggregationTemporality,
            collection_start_nano: int,
        ) -> _DataPointVarT | None:
            """
            Atomically return a point for the current value of the metric.
            """

            with self._lock:
                value = self._value
                sum_ = self._sum
                min_ = self._min
                max_ = self._max

                self._value = None
                self._sum = 0
                self._min = math.inf
                self._max = -math.inf

                if (
                    self._instrument_aggregation_temporality
                    is AggregationTemporality.DELTA
                ):
                    # This happens when the corresponding instrument for this
                    # aggregation is synchronous.
                    if (
                        collection_aggregation_temporality
                        is AggregationTemporality.DELTA
                    ):
                        previous_collection_start_nano = (
                            self._previous_collection_start_nano
                        )
                        self._previous_collection_start_nano = (
                            collection_start_nano
                        )

                        if value is None:
                            return None

                        return HistogramDataPoint(
                            attributes=self._attributes,
                            exemplars=self._collect_exemplars(),
                            start_time_unix_nano=previous_collection_start_nano,
                            time_unix_nano=collection_start_nano,
                            count=sum(value),
                            sum=sum_,
                            bucket_counts=tuple(value),
                            explicit_bounds=self._boundaries,
                            min=min_,
                            max=max_,
                        )

                    if value is None:
                        value = self._get_empty_bucket_counts()

                    if self._previous_value is None:
                        self._previous_value = self._get_empty_bucket_counts()

                    self._previous_value = [
                        value_element + previous_value_element
                        for (
                            value_element,
                            previous_value_element,
                        ) in zip(value, self._previous_value)
                    ]
                    self._previous_min = min(min_, self._previous_min)
                    self._previous_max = max(max_, self._previous_max)
                    self._previous_sum = sum_ + self._previous_sum

                    return HistogramDataPoint(
                        attributes=self._attributes,
                        exemplars=self._collect_exemplars(),
                        start_time_unix_nano=self._start_time_unix_nano,
                        time_unix_nano=collection_start_nano,
                        count=sum(self._previous_value),
                        sum=self._previous_sum,
                        bucket_counts=tuple(self._previous_value),
                        explicit_bounds=self._boundaries,
                        min=self._previous_min,
                        max=self._previous_max,
                    )

                return None

    # pylint: disable=protected-access
    class _ExponentialBucketHistogramAggregation(_Aggregation[HistogramPoint]):
        # _min_max_size and _max_max_size are the smallest and largest values
        # the max_size parameter may have, respectively.

        # _min_max_size is is the smallest reasonable value which is small enough
        # to contain the entire normal floating point range at the minimum scale.
        _min_max_size = 2

        # _max_max_size is an arbitrary limit meant to limit accidental creation of
        # giant exponential bucket histograms.
        _max_max_size = 16384

        def __init__(
            self,
            attributes: Attributes,
            reservoir_builder: ExemplarReservoirBuilder,
            instrument_aggregation_temporality: AggregationTemporality,
            start_time_unix_nano: int,
            # This is the default maximum number of buckets per positive or
            # negative number range.  The value 160 is specified by OpenTelemetry.
            # See the derivation here:
            # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#exponential-bucket-histogram-aggregation)
            max_size: int = 160,
            max_scale: int = 20,
            record_min_max: bool = True,
        ):
            # max_size is the maximum capacity of the positive and negative
            # buckets.
            # _sum is the sum of all the values aggregated by this aggregator.
            # _count is the count of all calls to aggregate.
            # _zero_count is the count of all the calls to aggregate when the value
            # to be aggregated is exactly 0.
            # _min is the smallest value aggregated by this aggregator.
            # _max is the smallest value aggregated by this aggregator.
            # _positive holds the positive values.
            # _negative holds the negative values by their absolute value.
            if max_size < self._min_max_size:
                raise ValueError(
                    f"Buckets max size {max_size} is smaller than "
                    "minimum max size {self._min_max_size}"
                )

            if max_size > self._max_max_size:
                raise ValueError(
                    f"Buckets max size {max_size} is larger than "
                    "maximum max size {self._max_max_size}"
                )
            if max_scale > 20:
                _logger.warning(
                    "max_scale is set to %s which is "
                    "larger than the recommended value of 20",
                    max_scale,
                )

            # This aggregation is analogous to _ExplicitBucketHistogramAggregation,
            # the only difference is that with every call to aggregate, the size
            # and amount of buckets can change (in
            # _ExplicitBucketHistogramAggregation both size and amount of buckets
            # remain constant once it is instantiated).

            super().__init__(
                attributes,
                reservoir_builder=partial(
                    reservoir_builder, size=min(20, max_size)
                ),
            )

            self._instrument_aggregation_temporality = (
                instrument_aggregation_temporality
            )
            self._start_time_unix_nano = start_time_unix_nano
            self._max_size = max_size
            self._max_scale = max_scale
            self._record_min_max = record_min_max

            self._value_positive = None
            self._value_negative = None
            self._min = math.inf
            self._max = -math.inf
            self._sum = 0
            self._count = 0
            self._zero_count = 0
            self._scale = None

            self._previous_value_positive = None
            self._previous_value_negative = None
            self._previous_min = math.inf
            self._previous_max = -math.inf
            self._previous_sum = 0
            self._previous_count = 0
            self._previous_zero_count = 0
            self._previous_scale = None

            self._previous_collection_start_nano = self._start_time_unix_nano

            self._mapping = self._new_mapping(self._max_scale)

        def aggregate(
            self, measurement: Measurement, should_sample_exemplar: bool = True
        ) -> None:
            # pylint: disable=too-many-branches,too-many-statements, too-many-locals

            with self._lock:
                if self._value_positive is None:
                    self._value_positive = Buckets()
                if self._value_negative is None:
                    self._value_negative = Buckets()

                measurement_value = measurement.value

                self._sum += measurement_value

                if self._record_min_max:
                    self._min = min(self._min, measurement_value)
                    self._max = max(self._max, measurement_value)

                self._count += 1

                if measurement_value == 0:
                    self._zero_count += 1

                    if self._count == self._zero_count:
                        self._scale = 0

                    return

                if measurement_value > 0:
                    value = self._value_positive

                else:
                    measurement_value = -measurement_value
                    value = self._value_negative

                # The following code finds out if it is necessary to change the
                # buckets to hold the incoming measurement_value, changes them if
                # necessary. This process does not exist in
                # _ExplicitBucketHistogram aggregation because the buckets there
                # are constant in size and amount.
                index = self._mapping.map_to_index(measurement_value)

                is_rescaling_needed = False
                low, high = 0, 0

                if len(value) == 0:
                    value.index_start = index
                    value.index_end = index
                    value.index_base = index

                elif (
                    index < value.index_start
                    and (value.index_end - index) >= self._max_size
                ):
                    is_rescaling_needed = True
                    low = index
                    high = value.index_end

                elif (
                    index > value.index_end
                    and (index - value.index_start) >= self._max_size
                ):
                    is_rescaling_needed = True
                    low = value.index_start
                    high = index

                if is_rescaling_needed:
                    scale_change = self._get_scale_change(low, high)
                    self._downscale(
                        scale_change,
                        self._value_positive,
                        self._value_negative,
                    )
                    self._mapping = self._new_mapping(
                        self._mapping.scale - scale_change
                    )

                    index = self._mapping.map_to_index(measurement_value)

                self._scale = self._mapping.scale

                if index < value.index_start:
                    span = value.index_end - index

                    if span >= len(value.counts):
                        value.grow(span + 1, self._max_size)

                    value.index_start = index

                elif index > value.index_end:
                    span = index - value.index_start

                    if span >= len(value.counts):
                        value.grow(span + 1, self._max_size)

                    value.index_end = index

                bucket_index = index - value.index_base

                if bucket_index < 0:
                    bucket_index += len(value.counts)

                # Now the buckets have been changed if needed and bucket_index will
                # be used to increment the counter of the bucket that needs to be
                # incremented.

                # This is analogous to
                # self._value[bisect_left(self._boundaries, measurement_value)] += 1
                # in _ExplicitBucketHistogramAggregation.aggregate
                value.increment_bucket(bucket_index)

            self._sample_exemplar(measurement, should_sample_exemplar)

        def collect(
            self,
            collection_aggregation_temporality: AggregationTemporality,
            collection_start_nano: int,
        ) -> _DataPointVarT | None:
            """
            Atomically return a point for the current value of the metric.
            """

            # pylint: disable=too-many-statements, too-many-locals
            with self._lock:
                value_positive = self._value_positive
                value_negative = self._value_negative
                sum_ = self._sum
                min_ = self._min
                max_ = self._max
                count = self._count
                zero_count = self._zero_count
                scale = self._scale

                self._value_positive = None
                self._value_negative = None
                self._sum = 0
                self._min = math.inf
                self._max = -math.inf
                self._count = 0
                self._zero_count = 0
                self._scale = None

                if (
                    self._instrument_aggregation_temporality
                    is AggregationTemporality.DELTA
                ):
                    # This happens when the corresponding instrument for this
                    # aggregation is synchronous.
                    if (
                        collection_aggregation_temporality
                        is AggregationTemporality.DELTA
                    ):
                        previous_collection_start_nano = (
                            self._previous_collection_start_nano
                        )
                        self._previous_collection_start_nano = (
                            collection_start_nano
                        )

                        if value_positive is None and value_negative is None:
                            return None

                        return ExponentialHistogramDataPoint(
                            attributes=self._attributes,
                            exemplars=self._collect_exemplars(),
                            start_time_unix_nano=previous_collection_start_nano,
                            time_unix_nano=collection_start_nano,
                            count=count,
                            sum=sum_,
                            scale=scale,
                            zero_count=zero_count,
                            positive=BucketsPoint(
                                offset=value_positive.offset,
                                bucket_counts=(value_positive.get_offset_counts()),
                            ),
                            negative=BucketsPoint(
                                offset=value_negative.offset,
                                bucket_counts=(value_negative.get_offset_counts()),
                            ),
                            # FIXME: Find the right value for flags
                            flags=0,
                            min=min_,
                            max=max_,
                        )

                    # Here collection_temporality is CUMULATIVE.
                    # instrument_temporality is always DELTA for the time being.
                    # Here we need to handle the case where:
                    # collect is called after at least one other call to collect
                    # (there is data in previous buckets, a call to merge is needed
                    # to handle possible differences in bucket sizes).
                    # collect is called without another call previous call to
                    # collect was made (there is no previous buckets, previous,
                    # empty buckets that are the same scale of the current buckets
                    # need to be made so that they can be cumulatively aggregated
                    # to the current buckets).

                    if (
                        value_positive is None
                        and self._previous_value_positive is None
                    ):
                        # This happens if collect is called for the first time
                        # and aggregate has not yet been called.
                        value_positive = Buckets()
                        self._previous_value_positive = value_positive.copy_empty()
                    if (
                        value_negative is None
                        and self._previous_value_negative is None
                    ):
                        value_negative = Buckets()
                        self._previous_value_negative = value_negative.copy_empty()
                    if scale is None and self._previous_scale is None:
                        scale = self._mapping.scale
                        self._previous_scale = scale

                    if (
                        value_positive is not None
                        and self._previous_value_positive is None
                    ):
                        # This happens when collect is called the very first time
                        # and aggregate has been called before.

                        # We need previous buckets to add them to the current ones.
                        # When collect is called for the first time, there are no
                        # previous buckets, so we need to create empty buckets to
                        # add them to the current ones. The addition of empty
                        # buckets to the current ones will result in the current
                        # ones unchanged.

                        # The way the previous buckets are generated here is
                        # different from the explicit bucket histogram where
                        # the size and amount of the buckets does not change once
                        # they are instantiated. Here, the size and amount of the
                        # buckets can change with every call to aggregate. In order
                        # to get empty buckets that can be added to the current
                        # ones resulting in the current ones unchanged we need to
                        # generate empty buckets that have the same size and amount
                        # as the current ones, this is what copy_empty does.
                        self._previous_value_positive = value_positive.copy_empty()
                    if (
                        value_negative is not None
                        and self._previous_value_negative is None
                    ):
                        self._previous_value_negative = value_negative.copy_empty()
                    if scale is not None and self._previous_scale is None:
                        self._previous_scale = scale

                    if (
                        value_positive is None
                        and self._previous_value_positive is not None
                    ):
                        value_positive = self._previous_value_positive.copy_empty()
                    if (
                        value_negative is None
                        and self._previous_value_negative is not None
                    ):
                        value_negative = self._previous_value_negative.copy_empty()
                    if scale is None and self._previous_scale is not None:
                        scale = self._previous_scale

                    # here self._previous_value_negative and self._previous_value_positive are not Optional anymore
                    self._previous_value_negative = cast(
                        Buckets, self._previous_value_negative
                    )
                    self._previous_value_positive = cast(
                        Buckets, self._previous_value_positive
                    )

                    min_scale = min(self._previous_scale, scale)

                    low_positive, high_positive = (
                        self._get_low_high_previous_current(
                            self._previous_value_positive,
                            value_positive,
                            scale,
                            min_scale,
                        )
                    )
                    low_negative, high_negative = (
                        self._get_low_high_previous_current(
                            self._previous_value_negative,
                            value_negative,
                            scale,
                            min_scale,
                        )
                    )

                    min_scale = min(
                        min_scale
                        - self._get_scale_change(low_positive, high_positive),
                        min_scale
                        - self._get_scale_change(low_negative, high_negative),
                    )

                    self._downscale(
                        self._previous_scale - min_scale,
                        self._previous_value_positive,
                        self._previous_value_negative,
                    )

                    # self._merge adds the values from value to
                    # self._previous_value, this is analogous to
                    # self._previous_value = [
                    #     value_element + previous_value_element
                    #     for (
                    #         value_element,
                    #         previous_value_element,
                    #     ) in zip(value, self._previous_value)
                    # ]
                    # in _ExplicitBucketHistogramAggregation.collect.
                    self._merge(
                        self._previous_value_positive,
                        value_positive,
                        scale,
                        min_scale,
                        collection_aggregation_temporality,
                    )
                    self._merge(
                        self._previous_value_negative,
                        value_negative,
                        scale,
                        min_scale,
                        collection_aggregation_temporality,
                    )

                    self._previous_min = min(min_, self._previous_min)
                    self._previous_max = max(max_, self._previous_max)
                    self._previous_sum = sum_ + self._previous_sum
                    self._previous_count = count + self._previous_count
                    self._previous_zero_count = (
                        zero_count + self._previous_zero_count
                    )
                    self._previous_scale = min_scale

                    return ExponentialHistogramDataPoint(
                        attributes=self._attributes,
                        exemplars=self._collect_exemplars(),
                        start_time_unix_nano=self._start_time_unix_nano,
                        time_unix_nano=collection_start_nano,
                        count=self._previous_count,
                        sum=self._previous_sum,
                        scale=self._previous_scale,
                        zero_count=self._previous_zero_count,
                        positive=BucketsPoint(
                            offset=self._previous_value_positive.offset,
                            bucket_counts=(
                                self._previous_value_positive.get_offset_counts()
                            ),
                        ),
                        negative=BucketsPoint(
                            offset=self._previous_value_negative.offset,
                            bucket_counts=(
                                self._previous_value_negative.get_offset_counts()
                            ),
                        ),
                        # FIXME: Find the right value for flags
                        flags=0,
                        min=self._previous_min,
                        max=self._previous_max,
                    )

                return None

        def _get_low_high_previous_current(
            self,
            previous_point_buckets,
            current_point_buckets,
            current_scale,
            min_scale,
        ):
            (previous_point_low, previous_point_high) = self._get_low_high(
                previous_point_buckets, self._previous_scale, min_scale
            )
            (current_point_low, current_point_high) = self._get_low_high(
                current_point_buckets, current_scale, min_scale
            )

            if current_point_low > current_point_high:
                low = previous_point_low
                high = previous_point_high

            elif previous_point_low > previous_point_high:
                low = current_point_low
                high = current_point_high

            else:
                low = min(previous_point_low, current_point_low)
                high = max(previous_point_high, current_point_high)

            return low, high

        @staticmethod
        def _get_low_high(buckets, scale, min_scale):
            if buckets.counts == [0]:
                return 0, -1

            shift = scale - min_scale

            return buckets.index_start >> shift, buckets.index_end >> shift

        @staticmethod
        def _new_mapping(scale: int) -> Mapping:
            if scale <= 0:
                return ExponentMapping(scale)
            return LogarithmMapping(scale)

        def _get_scale_change(self, low, high):
            change = 0

            while high - low >= self._max_size:
                high = high >> 1
                low = low >> 1

                change += 1

            return change

        @staticmethod
        def _downscale(change: int, positive, negative):
            if change == 0:
                return

            if change < 0:
                # pylint: disable=broad-exception-raised
                raise Exception("Invalid change of scale")

            positive.downscale(change)
            negative.downscale(change)

        def _merge(
            self,
            previous_buckets: Buckets,
            current_buckets: Buckets,
            current_scale,
            min_scale,
            aggregation_temporality,
        ):
            current_change = current_scale - min_scale

            for current_bucket_index, current_bucket in enumerate(
                current_buckets.counts
            ):
                if current_bucket == 0:
                    continue

                # Not considering the case where len(previous_buckets) == 0. This
                # would not happen because self._previous_point is only assigned to
                # an ExponentialHistogramDataPoint object if self._count != 0.

                current_index = current_buckets.index_base + current_bucket_index
                if current_index > current_buckets.index_end:
                    current_index -= len(current_buckets.counts)

                index = current_index >> current_change

                if index < previous_buckets.index_start:
                    span = previous_buckets.index_end - index

                    if span >= self._max_size:
                        # pylint: disable=broad-exception-raised
                        raise Exception("Incorrect merge scale")

                    if span >= len(previous_buckets.counts):
                        previous_buckets.grow(span + 1, self._max_size)

                    previous_buckets.index_start = index

                if index > previous_buckets.index_end:
                    span = index - previous_buckets.index_start

                    if span >= self._max_size:
                        # pylint: disable=broad-exception-raised
                        raise Exception("Incorrect merge scale")

                    if span >= len(previous_buckets.counts):
                        previous_buckets.grow(span + 1, self._max_size)

                    previous_buckets.index_end = index

                bucket_index = index - previous_buckets.index_base

                if bucket_index < 0:
                    bucket_index += len(previous_buckets.counts)

                if aggregation_temporality is AggregationTemporality.DELTA:
                    current_bucket = -current_bucket

                previous_buckets.increment_bucket(
                    bucket_index, increment=current_bucket
                )

    class Aggregation(ABC):
        """
        Base class for all aggregation types.
        """

        @abstractmethod
        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            """Creates an aggregation"""

    class DefaultAggregation(Aggregation):
        """
        The default aggregation to be used in a `View`.

        This aggregation will create an actual aggregation depending on the
        instrument type, as specified next:

        ==================================================== ====================================
        Instrument                                           Aggregation
        ==================================================== ====================================
        `opentelemetry.sdk.metrics.Counter`                  `SumAggregation`
        `opentelemetry.sdk.metrics.UpDownCounter`            `SumAggregation`
        `opentelemetry.sdk.metrics.ObservableCounter`        `SumAggregation`
        `opentelemetry.sdk.metrics.ObservableUpDownCounter`  `SumAggregation`
        `opentelemetry.sdk.metrics.Histogram`                `ExplicitBucketHistogramAggregation`
        `opentelemetry.sdk.metrics.ObservableGauge`          `LastValueAggregation`
        ==================================================== ====================================
        """

        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            # pylint: disable=too-many-return-statements
            if isinstance(instrument, Counter):
                return _SumAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(_SumAggregation),
                    instrument_is_monotonic=True,
                    instrument_aggregation_temporality=(
                        AggregationTemporality.DELTA
                    ),
                    start_time_unix_nano=start_time_unix_nano,
                )
            if isinstance(instrument, UpDownCounter):
                return _SumAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(_SumAggregation),
                    instrument_is_monotonic=False,
                    instrument_aggregation_temporality=(
                        AggregationTemporality.DELTA
                    ),
                    start_time_unix_nano=start_time_unix_nano,
                )

            if isinstance(instrument, ObservableCounter):
                return _SumAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(_SumAggregation),
                    instrument_is_monotonic=True,
                    instrument_aggregation_temporality=(
                        AggregationTemporality.CUMULATIVE
                    ),
                    start_time_unix_nano=start_time_unix_nano,
                )

            if isinstance(instrument, ObservableUpDownCounter):
                return _SumAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(_SumAggregation),
                    instrument_is_monotonic=False,
                    instrument_aggregation_temporality=(
                        AggregationTemporality.CUMULATIVE
                    ),
                    start_time_unix_nano=start_time_unix_nano,
                )

            if isinstance(instrument, Histogram):
                boundaries = instrument._advisory.explicit_bucket_boundaries
                return _ExplicitBucketHistogramAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(
                        _ExplicitBucketHistogramAggregation
                    ),
                    instrument_aggregation_temporality=(
                        AggregationTemporality.DELTA
                    ),
                    boundaries=boundaries,
                    start_time_unix_nano=start_time_unix_nano,
                )

            if isinstance(instrument, ObservableGauge):
                return _LastValueAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(_LastValueAggregation),
                )

            if isinstance(instrument, _Gauge):
                return _LastValueAggregation(
                    attributes,
                    reservoir_builder=reservoir_factory(_LastValueAggregation),
                )

            # pylint: disable=broad-exception-raised
            raise Exception(f"Invalid instrument type {type(instrument)} found")

    class ExponentialBucketHistogramAggregation(Aggregation):
        """This aggregation informs the SDK to collect:

        - Count of Measurement values falling using a base-2 exponential formula.
        - Arithmetic sum of Measurement values in population. This SHOULD NOT be collected when used with instruments that record negative measurements, e.g. UpDownCounter or ObservableGauge.
        - Min (optional) Measurement value in population.
        - Max (optional) Measurement value in population.

        Args:
            max_size: Maximum number of buckets in each of the positive and negative ranges, not counting the special zero bucket.
            max_scale: Maximum scale factor.
            record_min_max: Whether to record min and max.
        """

        def __init__(
            self,
            max_size: int = 160,
            max_scale: int = 20,
            record_min_max: bool = True,
        ):
            self._max_size = max_size
            self._max_scale = max_scale
            self._record_min_max = record_min_max

        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            instrument_aggregation_temporality = AggregationTemporality.UNSPECIFIED
            if isinstance(instrument, Synchronous):
                instrument_aggregation_temporality = AggregationTemporality.DELTA
            elif isinstance(instrument, Asynchronous):
                instrument_aggregation_temporality = (
                    AggregationTemporality.CUMULATIVE
                )

            return _ExponentialBucketHistogramAggregation(
                attributes,
                reservoir_factory(_ExponentialBucketHistogramAggregation),
                instrument_aggregation_temporality,
                start_time_unix_nano,
                max_size=self._max_size,
                max_scale=self._max_scale,
                record_min_max=self._record_min_max,
            )

    class ExplicitBucketHistogramAggregation(Aggregation):
        """This aggregation informs the SDK to collect:

        - Count of Measurement values falling within explicit bucket boundaries.
        - Arithmetic sum of Measurement values in population. This SHOULD NOT be collected when used with instruments that record negative measurements, e.g. UpDownCounter or ObservableGauge.
        - Min (optional) Measurement value in population.
        - Max (optional) Measurement value in population.

        Args:
            boundaries: Array of increasing values representing explicit bucket boundary values.
            record_min_max: Whether to record min and max.
        """

        def __init__(
            self,
            boundaries: Sequence[float] | None = None,
            record_min_max: bool = True,
        ) -> None:
            self._boundaries = boundaries
            self._record_min_max = record_min_max

        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            instrument_aggregation_temporality = AggregationTemporality.UNSPECIFIED
            if isinstance(instrument, Synchronous):
                instrument_aggregation_temporality = AggregationTemporality.DELTA
            elif isinstance(instrument, Asynchronous):
                instrument_aggregation_temporality = (
                    AggregationTemporality.CUMULATIVE
                )

            if self._boundaries is not None:
                boundaries = self._boundaries
            else:
                # guard for usage with instruments without advisory
                advisory = getattr(instrument, "_advisory", None)
                boundaries = (
                    advisory.explicit_bucket_boundaries
                    if advisory is not None
                    else None
                )

            return _ExplicitBucketHistogramAggregation(
                attributes,
                instrument_aggregation_temporality,
                start_time_unix_nano,
                reservoir_factory(_ExplicitBucketHistogramAggregation),
                boundaries,
                self._record_min_max,
            )

    class SumAggregation(Aggregation):
        """This aggregation informs the SDK to collect:

        - The arithmetic sum of Measurement values.
        """

        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            instrument_aggregation_temporality = AggregationTemporality.UNSPECIFIED
            if isinstance(instrument, Synchronous):
                instrument_aggregation_temporality = AggregationTemporality.DELTA
            elif isinstance(instrument, Asynchronous):
                instrument_aggregation_temporality = (
                    AggregationTemporality.CUMULATIVE
                )

            return _SumAggregation(
                attributes,
                isinstance(instrument, (Counter, ObservableCounter)),
                instrument_aggregation_temporality,
                start_time_unix_nano,
                reservoir_factory(_SumAggregation),
            )

    class LastValueAggregation(Aggregation):
        """
        This aggregation informs the SDK to collect:

        - The last Measurement.
        - The timestamp of the last Measurement.
        """

        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            return _LastValueAggregation(
                attributes,
                reservoir_builder=reservoir_factory(_LastValueAggregation),
            )

    class DropAggregation(Aggregation):
        """Using this aggregation will make all measurements be ignored."""

        def _create_aggregation(
            self,
            instrument: _Instrument,
            attributes: Attributes,
            reservoir_factory: Callable[
                [type[_Aggregation]], ExemplarReservoirBuilder
            ],
            start_time_unix_nano: int,
        ) -> _Aggregation:
            return _DropAggregation(
                attributes, reservoir_factory(_DropAggregation)
            )