We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Implement temporality conversion
Blocked by #2327

Temporality conversion is specified in the [design doc](https://docs.google.com/document/d/1FOmpGCiZAWTd6a3vcEHBnm1LKp7HAKuD4BBN3YCkymc/edit#heading=h.ulghopzfw7ou). For this issue:

- Create a component to handle temporality conversion and storing of previous collection interval's cumulative point value when necessary. For example, there is no need to store previous cumulative for async instruments and cumulative export temporality.
- Write temporality conversion algorithm. Use the export format added for #2327 as the input and output of the algorithm

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from abc import ABC, abstractmethod
16 from bisect import bisect_left
17 from logging import getLogger
18 from math import inf
19 from threading import Lock
20 from typing import Generic, List, Optional, Sequence, TypeVar
21 
22 from opentelemetry.sdk._metrics.measurement import Measurement
23 from opentelemetry.sdk._metrics.point import (
24     AggregationTemporality,
25     Gauge,
26     Histogram,
27     PointT,
28     Sum,
29 )
30 from opentelemetry.util._time import _time_ns
31 
32 _PointVarT = TypeVar("_PointVarT", bound=PointT)
33 
34 _logger = getLogger(__name__)
35 
36 
37 class _InstrumentMonotonicityAwareAggregation:
38     def __init__(self, instrument_is_monotonic: bool):
39         self._instrument_is_monotonic = instrument_is_monotonic
40         super().__init__()
41 
42 
43 class Aggregation(ABC, Generic[_PointVarT]):
44     def __init__(self):
45         self._lock = Lock()
46 
47     @abstractmethod
48     def aggregate(self, measurement: Measurement) -> None:
49         pass
50 
51     @abstractmethod
52     def collect(self) -> Optional[_PointVarT]:
53         pass
54 
55 
56 class SynchronousSumAggregation(
57     _InstrumentMonotonicityAwareAggregation, Aggregation[Sum]
58 ):
59     def __init__(self, instrument_is_monotonic: bool):
60         super().__init__(instrument_is_monotonic)
61         self._value = 0
62         self._start_time_unix_nano = _time_ns()
63 
64     def aggregate(self, measurement: Measurement) -> None:
65         with self._lock:
66             self._value = self._value + measurement.value
67 
68     def collect(self) -> Optional[Sum]:
69         """
70         Atomically return a point for the current value of the metric and
71         reset the aggregation value.
72         """
73         now = _time_ns()
74 
75         with self._lock:
76             value = self._value
77             start_time_unix_nano = self._start_time_unix_nano
78 
79             self._value = 0
80             self._start_time_unix_nano = now + 1
81 
82         return Sum(
83             aggregation_temporality=AggregationTemporality.DELTA,
84             is_monotonic=self._instrument_is_monotonic,
85             start_time_unix_nano=start_time_unix_nano,
86             time_unix_nano=now,
87             value=value,
88         )
89 
90 
91 class AsynchronousSumAggregation(
92     _InstrumentMonotonicityAwareAggregation, Aggregation[Sum]
93 ):
94     def __init__(self, instrument_is_monotonic: bool):
95         super().__init__(instrument_is_monotonic)
96         self._value = None
97         self._start_time_unix_nano = _time_ns()
98 
99     def aggregate(self, measurement: Measurement) -> None:
100         with self._lock:
101             self._value = measurement.value
102 
103     def collect(self) -> Optional[Sum]:
104         """
105         Atomically return a point for the current value of the metric.
106         """
107         if self._value is None:
108             return None
109 
110         return Sum(
111             start_time_unix_nano=self._start_time_unix_nano,
112             time_unix_nano=_time_ns(),
113             value=self._value,
114             aggregation_temporality=AggregationTemporality.CUMULATIVE,
115             is_monotonic=self._instrument_is_monotonic,
116         )
117 
118 
119 class LastValueAggregation(Aggregation[Gauge]):
120     def __init__(self):
121         super().__init__()
122         self._value = None
123 
124     def aggregate(self, measurement: Measurement):
125         with self._lock:
126             self._value = measurement.value
127 
128     def collect(self) -> Optional[Gauge]:
129         """
130         Atomically return a point for the current value of the metric.
131         """
132         if self._value is None:
133             return None
134 
135         return Gauge(
136             time_unix_nano=_time_ns(),
137             value=self._value,
138         )
139 
140 
141 class ExplicitBucketHistogramAggregation(Aggregation[Histogram]):
142     def __init__(
143         self,
144         boundaries: Sequence[float] = (
145             0.0,
146             5.0,
147             10.0,
148             25.0,
149             50.0,
150             75.0,
151             100.0,
152             250.0,
153             500.0,
154             1000.0,
155         ),
156         record_min_max: bool = True,
157     ):
158         super().__init__()
159         self._boundaries = tuple(boundaries)
160         self._bucket_counts = self._get_empty_bucket_counts()
161         self._min = inf
162         self._max = -inf
163         self._sum = 0
164         self._record_min_max = record_min_max
165         self._start_time_unix_nano = _time_ns()
166 
167     def _get_empty_bucket_counts(self) -> List[int]:
168         return [0] * (len(self._boundaries) + 1)
169 
170     def aggregate(self, measurement: Measurement) -> None:
171 
172         value = measurement.value
173 
174         if self._record_min_max:
175             self._min = min(self._min, value)
176             self._max = max(self._max, value)
177 
178         self._sum += value
179 
180         self._bucket_counts[bisect_left(self._boundaries, value)] += 1
181 
182     def collect(self) -> Optional[Histogram]:
183         """
184         Atomically return a point for the current value of the metric.
185         """
186         now = _time_ns()
187 
188         with self._lock:
189             value = self._bucket_counts
190             start_time_unix_nano = self._start_time_unix_nano
191 
192             self._bucket_counts = self._get_empty_bucket_counts()
193             self._start_time_unix_nano = now + 1
194 
195         return Histogram(
196             start_time_unix_nano=start_time_unix_nano,
197             time_unix_nano=now,
198             bucket_counts=tuple(value),
199             explicit_bounds=self._boundaries,
200             aggregation_temporality=AggregationTemporality.DELTA,
201             sum=self._sum,
202         )
203
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py
--- a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py
@@ -14,6 +14,7 @@
 
 from abc import ABC, abstractmethod
 from bisect import bisect_left
+from dataclasses import replace
 from logging import getLogger
 from math import inf
 from threading import Lock
@@ -200,3 +201,80 @@
             aggregation_temporality=AggregationTemporality.DELTA,
             sum=self._sum,
         )
+
+
+def _convert_aggregation_temporality(
+    previous_point: Optional[_PointVarT],
+    current_point: _PointVarT,
+    aggregation_temporality: AggregationTemporality,
+) -> _PointVarT:
+    """Converts `current_point` to the requested `aggregation_temporality`
+    given the `previous_point`.
+
+    `previous_point` must have `CUMULATIVE` temporality. `current_point` may
+    have `DELTA` or `CUMULATIVE` temporality.
+
+    The output point will have temporality `aggregation_temporality`. Since
+    `GAUGE` points have no temporality, they are returned unchanged.
+    """
+
+    current_point_type = type(current_point)
+
+    if current_point_type is Gauge:
+        return current_point
+
+    if previous_point is not None and type(previous_point) is not type(
+        current_point
+    ):
+        _logger.warning(
+            "convert_aggregation_temporality called with mismatched "
+            "point types: %s and %s",
+            type(previous_point),
+            current_point_type,
+        )
+
+        return current_point
+
+    if current_point_type is Sum:
+        if previous_point is None:
+            # Output CUMULATIVE for a synchronous instrument
+            # There is no previous value, return the delta point as a
+            # cumulative
+            return replace(
+                current_point, aggregation_temporality=aggregation_temporality
+            )
+        if previous_point.aggregation_temporality is not (
+            AggregationTemporality.CUMULATIVE
+        ):
+            raise Exception(
+                "previous_point aggregation temporality must be CUMULATIVE"
+            )
+
+        if current_point.aggregation_temporality is aggregation_temporality:
+            # Output DELTA for a synchronous instrument
+            # Output CUMULATIVE for an asynchronous instrument
+            return current_point
+
+        if aggregation_temporality is AggregationTemporality.DELTA:
+            # Output temporality DELTA for an asynchronous instrument
+            value = current_point.value - previous_point.value
+            output_start_time_unix_nano = previous_point.time_unix_nano
+
+        else:
+            # Output CUMULATIVE for a synchronous instrument
+            value = current_point.value + previous_point.value
+            output_start_time_unix_nano = previous_point.start_time_unix_nano
+
+        is_monotonic = (
+            previous_point.is_monotonic and current_point.is_monotonic
+        )
+
+        return Sum(
+            start_time_unix_nano=output_start_time_unix_nano,
+            time_unix_nano=current_point.time_unix_nano,
+            value=value,
+            aggregation_temporality=aggregation_temporality,
+            is_monotonic=is_monotonic,
+        )
+
+    return None

{"golden_diff": "diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py\n--- a/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py\n+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py\n@@ -14,6 +14,7 @@\n \n from abc import ABC, abstractmethod\n from bisect import bisect_left\n+from dataclasses import replace\n from logging import getLogger\n from math import inf\n from threading import Lock\n@@ -200,3 +201,80 @@\n             aggregation_temporality=AggregationTemporality.DELTA,\n             sum=self._sum,\n         )\n+\n+\n+def _convert_aggregation_temporality(\n+    previous_point: Optional[_PointVarT],\n+    current_point: _PointVarT,\n+    aggregation_temporality: AggregationTemporality,\n+) -> _PointVarT:\n+    \"\"\"Converts `current_point` to the requested `aggregation_temporality`\n+    given the `previous_point`.\n+\n+    `previous_point` must have `CUMULATIVE` temporality. `current_point` may\n+    have `DELTA` or `CUMULATIVE` temporality.\n+\n+    The output point will have temporality `aggregation_temporality`. Since\n+    `GAUGE` points have no temporality, they are returned unchanged.\n+    \"\"\"\n+\n+    current_point_type = type(current_point)\n+\n+    if current_point_type is Gauge:\n+        return current_point\n+\n+    if previous_point is not None and type(previous_point) is not type(\n+        current_point\n+    ):\n+        _logger.warning(\n+            \"convert_aggregation_temporality called with mismatched \"\n+            \"point types: %s and %s\",\n+            type(previous_point),\n+            current_point_type,\n+        )\n+\n+        return current_point\n+\n+    if current_point_type is Sum:\n+        if previous_point is None:\n+            # Output CUMULATIVE for a synchronous instrument\n+            # There is no previous value, return the delta point as a\n+            # cumulative\n+            return replace(\n+                current_point, aggregation_temporality=aggregation_temporality\n+            )\n+        if previous_point.aggregation_temporality is not (\n+            AggregationTemporality.CUMULATIVE\n+        ):\n+            raise Exception(\n+                \"previous_point aggregation temporality must be CUMULATIVE\"\n+            )\n+\n+        if current_point.aggregation_temporality is aggregation_temporality:\n+            # Output DELTA for a synchronous instrument\n+            # Output CUMULATIVE for an asynchronous instrument\n+            return current_point\n+\n+        if aggregation_temporality is AggregationTemporality.DELTA:\n+            # Output temporality DELTA for an asynchronous instrument\n+            value = current_point.value - previous_point.value\n+            output_start_time_unix_nano = previous_point.time_unix_nano\n+\n+        else:\n+            # Output CUMULATIVE for a synchronous instrument\n+            value = current_point.value + previous_point.value\n+            output_start_time_unix_nano = previous_point.start_time_unix_nano\n+\n+        is_monotonic = (\n+            previous_point.is_monotonic and current_point.is_monotonic\n+        )\n+\n+        return Sum(\n+            start_time_unix_nano=output_start_time_unix_nano,\n+            time_unix_nano=current_point.time_unix_nano,\n+            value=value,\n+            aggregation_temporality=aggregation_temporality,\n+            is_monotonic=is_monotonic,\n+        )\n+\n+    return None\n", "issue": "Implement temporality conversion\nBlocked by #2327\r\n\r\nTemporality conversion is specified in the [design doc](https://docs.google.com/document/d/1FOmpGCiZAWTd6a3vcEHBnm1LKp7HAKuD4BBN3YCkymc/edit#heading=h.ulghopzfw7ou). For this issue:\r\n\r\n- Create a component to handle temporality conversion and storing of previous collection interval's cumulative point value when necessary. For example, there is no need to store previous cumulative for async instruments and cumulative export temporality.\r\n- Write temporality conversion algorithm. Use the export format added for #2327 as the input and output of the algorithm\n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom abc import ABC, abstractmethod\nfrom bisect import bisect_left\nfrom logging import getLogger\nfrom math import inf\nfrom threading import Lock\nfrom typing import Generic, List, Optional, Sequence, TypeVar\n\nfrom opentelemetry.sdk._metrics.measurement import Measurement\nfrom opentelemetry.sdk._metrics.point import (\n    AggregationTemporality,\n    Gauge,\n    Histogram,\n    PointT,\n    Sum,\n)\nfrom opentelemetry.util._time import _time_ns\n\n_PointVarT = TypeVar(\"_PointVarT\", bound=PointT)\n\n_logger = getLogger(__name__)\n\n\nclass _InstrumentMonotonicityAwareAggregation:\n    def __init__(self, instrument_is_monotonic: bool):\n        self._instrument_is_monotonic = instrument_is_monotonic\n        super().__init__()\n\n\nclass Aggregation(ABC, Generic[_PointVarT]):\n    def __init__(self):\n        self._lock = Lock()\n\n    @abstractmethod\n    def aggregate(self, measurement: Measurement) -> None:\n        pass\n\n    @abstractmethod\n    def collect(self) -> Optional[_PointVarT]:\n        pass\n\n\nclass SynchronousSumAggregation(\n    _InstrumentMonotonicityAwareAggregation, Aggregation[Sum]\n):\n    def __init__(self, instrument_is_monotonic: bool):\n        super().__init__(instrument_is_monotonic)\n        self._value = 0\n        self._start_time_unix_nano = _time_ns()\n\n    def aggregate(self, measurement: Measurement) -> None:\n        with self._lock:\n            self._value = self._value + measurement.value\n\n    def collect(self) -> Optional[Sum]:\n        \"\"\"\n        Atomically return a point for the current value of the metric and\n        reset the aggregation value.\n        \"\"\"\n        now = _time_ns()\n\n        with self._lock:\n            value = self._value\n            start_time_unix_nano = self._start_time_unix_nano\n\n            self._value = 0\n            self._start_time_unix_nano = now + 1\n\n        return Sum(\n            aggregation_temporality=AggregationTemporality.DELTA,\n            is_monotonic=self._instrument_is_monotonic,\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=now,\n            value=value,\n        )\n\n\nclass AsynchronousSumAggregation(\n    _InstrumentMonotonicityAwareAggregation, Aggregation[Sum]\n):\n    def __init__(self, instrument_is_monotonic: bool):\n        super().__init__(instrument_is_monotonic)\n        self._value = None\n        self._start_time_unix_nano = _time_ns()\n\n    def aggregate(self, measurement: Measurement) -> None:\n        with self._lock:\n            self._value = measurement.value\n\n    def collect(self) -> Optional[Sum]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        if self._value is None:\n            return None\n\n        return Sum(\n            start_time_unix_nano=self._start_time_unix_nano,\n            time_unix_nano=_time_ns(),\n            value=self._value,\n            aggregation_temporality=AggregationTemporality.CUMULATIVE,\n            is_monotonic=self._instrument_is_monotonic,\n        )\n\n\nclass LastValueAggregation(Aggregation[Gauge]):\n    def __init__(self):\n        super().__init__()\n        self._value = None\n\n    def aggregate(self, measurement: Measurement):\n        with self._lock:\n            self._value = measurement.value\n\n    def collect(self) -> Optional[Gauge]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        if self._value is None:\n            return None\n\n        return Gauge(\n            time_unix_nano=_time_ns(),\n            value=self._value,\n        )\n\n\nclass ExplicitBucketHistogramAggregation(Aggregation[Histogram]):\n    def __init__(\n        self,\n        boundaries: Sequence[float] = (\n            0.0,\n            5.0,\n            10.0,\n            25.0,\n            50.0,\n            75.0,\n            100.0,\n            250.0,\n            500.0,\n            1000.0,\n        ),\n        record_min_max: bool = True,\n    ):\n        super().__init__()\n        self._boundaries = tuple(boundaries)\n        self._bucket_counts = self._get_empty_bucket_counts()\n        self._min = inf\n        self._max = -inf\n        self._sum = 0\n        self._record_min_max = record_min_max\n        self._start_time_unix_nano = _time_ns()\n\n    def _get_empty_bucket_counts(self) -> List[int]:\n        return [0] * (len(self._boundaries) + 1)\n\n    def aggregate(self, measurement: Measurement) -> None:\n\n        value = measurement.value\n\n        if self._record_min_max:\n            self._min = min(self._min, value)\n            self._max = max(self._max, value)\n\n        self._sum += value\n\n        self._bucket_counts[bisect_left(self._boundaries, value)] += 1\n\n    def collect(self) -> Optional[Histogram]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        now = _time_ns()\n\n        with self._lock:\n            value = self._bucket_counts\n            start_time_unix_nano = self._start_time_unix_nano\n\n            self._bucket_counts = self._get_empty_bucket_counts()\n            self._start_time_unix_nano = now + 1\n\n        return Histogram(\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=now,\n            bucket_counts=tuple(value),\n            explicit_bounds=self._boundaries,\n            aggregation_temporality=AggregationTemporality.DELTA,\n            sum=self._sum,\n        )\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom abc import ABC, abstractmethod\nfrom bisect import bisect_left\nfrom dataclasses import replace\nfrom logging import getLogger\nfrom math import inf\nfrom threading import Lock\nfrom typing import Generic, List, Optional, Sequence, TypeVar\n\nfrom opentelemetry.sdk._metrics.measurement import Measurement\nfrom opentelemetry.sdk._metrics.point import (\n    AggregationTemporality,\n    Gauge,\n    Histogram,\n    PointT,\n    Sum,\n)\nfrom opentelemetry.util._time import _time_ns\n\n_PointVarT = TypeVar(\"_PointVarT\", bound=PointT)\n\n_logger = getLogger(__name__)\n\n\nclass _InstrumentMonotonicityAwareAggregation:\n    def __init__(self, instrument_is_monotonic: bool):\n        self._instrument_is_monotonic = instrument_is_monotonic\n        super().__init__()\n\n\nclass Aggregation(ABC, Generic[_PointVarT]):\n    def __init__(self):\n        self._lock = Lock()\n\n    @abstractmethod\n    def aggregate(self, measurement: Measurement) -> None:\n        pass\n\n    @abstractmethod\n    def collect(self) -> Optional[_PointVarT]:\n        pass\n\n\nclass SynchronousSumAggregation(\n    _InstrumentMonotonicityAwareAggregation, Aggregation[Sum]\n):\n    def __init__(self, instrument_is_monotonic: bool):\n        super().__init__(instrument_is_monotonic)\n        self._value = 0\n        self._start_time_unix_nano = _time_ns()\n\n    def aggregate(self, measurement: Measurement) -> None:\n        with self._lock:\n            self._value = self._value + measurement.value\n\n    def collect(self) -> Optional[Sum]:\n        \"\"\"\n        Atomically return a point for the current value of the metric and\n        reset the aggregation value.\n        \"\"\"\n        now = _time_ns()\n\n        with self._lock:\n            value = self._value\n            start_time_unix_nano = self._start_time_unix_nano\n\n            self._value = 0\n            self._start_time_unix_nano = now + 1\n\n        return Sum(\n            aggregation_temporality=AggregationTemporality.DELTA,\n            is_monotonic=self._instrument_is_monotonic,\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=now,\n            value=value,\n        )\n\n\nclass AsynchronousSumAggregation(\n    _InstrumentMonotonicityAwareAggregation, Aggregation[Sum]\n):\n    def __init__(self, instrument_is_monotonic: bool):\n        super().__init__(instrument_is_monotonic)\n        self._value = None\n        self._start_time_unix_nano = _time_ns()\n\n    def aggregate(self, measurement: Measurement) -> None:\n        with self._lock:\n            self._value = measurement.value\n\n    def collect(self) -> Optional[Sum]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        if self._value is None:\n            return None\n\n        return Sum(\n            start_time_unix_nano=self._start_time_unix_nano,\n            time_unix_nano=_time_ns(),\n            value=self._value,\n            aggregation_temporality=AggregationTemporality.CUMULATIVE,\n            is_monotonic=self._instrument_is_monotonic,\n        )\n\n\nclass LastValueAggregation(Aggregation[Gauge]):\n    def __init__(self):\n        super().__init__()\n        self._value = None\n\n    def aggregate(self, measurement: Measurement):\n        with self._lock:\n            self._value = measurement.value\n\n    def collect(self) -> Optional[Gauge]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        if self._value is None:\n            return None\n\n        return Gauge(\n            time_unix_nano=_time_ns(),\n            value=self._value,\n        )\n\n\nclass ExplicitBucketHistogramAggregation(Aggregation[Histogram]):\n    def __init__(\n        self,\n        boundaries: Sequence[float] = (\n            0.0,\n            5.0,\n            10.0,\n            25.0,\n            50.0,\n            75.0,\n            100.0,\n            250.0,\n            500.0,\n            1000.0,\n        ),\n        record_min_max: bool = True,\n    ):\n        super().__init__()\n        self._boundaries = tuple(boundaries)\n        self._bucket_counts = self._get_empty_bucket_counts()\n        self._min = inf\n        self._max = -inf\n        self._sum = 0\n        self._record_min_max = record_min_max\n        self._start_time_unix_nano = _time_ns()\n\n    def _get_empty_bucket_counts(self) -> List[int]:\n        return [0] * (len(self._boundaries) + 1)\n\n    def aggregate(self, measurement: Measurement) -> None:\n\n        value = measurement.value\n\n        if self._record_min_max:\n            self._min = min(self._min, value)\n            self._max = max(self._max, value)\n\n        self._sum += value\n\n        self._bucket_counts[bisect_left(self._boundaries, value)] += 1\n\n    def collect(self) -> Optional[Histogram]:\n        \"\"\"\n        Atomically return a point for the current value of the metric.\n        \"\"\"\n        now = _time_ns()\n\n        with self._lock:\n            value = self._bucket_counts\n            start_time_unix_nano = self._start_time_unix_nano\n\n            self._bucket_counts = self._get_empty_bucket_counts()\n            self._start_time_unix_nano = now + 1\n\n        return Histogram(\n            start_time_unix_nano=start_time_unix_nano,\n            time_unix_nano=now,\n            bucket_counts=tuple(value),\n            explicit_bounds=self._boundaries,\n            aggregation_temporality=AggregationTemporality.DELTA,\n            sum=self._sum,\n        )\n\n\ndef _convert_aggregation_temporality(\n    previous_point: Optional[_PointVarT],\n    current_point: _PointVarT,\n    aggregation_temporality: AggregationTemporality,\n) -> _PointVarT:\n    \"\"\"Converts `current_point` to the requested `aggregation_temporality`\n    given the `previous_point`.\n\n    `previous_point` must have `CUMULATIVE` temporality. `current_point` may\n    have `DELTA` or `CUMULATIVE` temporality.\n\n    The output point will have temporality `aggregation_temporality`. Since\n    `GAUGE` points have no temporality, they are returned unchanged.\n    \"\"\"\n\n    current_point_type = type(current_point)\n\n    if current_point_type is Gauge:\n        return current_point\n\n    if previous_point is not None and type(previous_point) is not type(\n        current_point\n    ):\n        _logger.warning(\n            \"convert_aggregation_temporality called with mismatched \"\n            \"point types: %s and %s\",\n            type(previous_point),\n            current_point_type,\n        )\n\n        return current_point\n\n    if current_point_type is Sum:\n        if previous_point is None:\n            # Output CUMULATIVE for a synchronous instrument\n            # There is no previous value, return the delta point as a\n            # cumulative\n            return replace(\n                current_point, aggregation_temporality=aggregation_temporality\n            )\n        if previous_point.aggregation_temporality is not (\n            AggregationTemporality.CUMULATIVE\n        ):\n            raise Exception(\n                \"previous_point aggregation temporality must be CUMULATIVE\"\n            )\n\n        if current_point.aggregation_temporality is aggregation_temporality:\n            # Output DELTA for a synchronous instrument\n            # Output CUMULATIVE for an asynchronous instrument\n            return current_point\n\n        if aggregation_temporality is AggregationTemporality.DELTA:\n            # Output temporality DELTA for an asynchronous instrument\n            value = current_point.value - previous_point.value\n            output_start_time_unix_nano = previous_point.time_unix_nano\n\n        else:\n            # Output CUMULATIVE for a synchronous instrument\n            value = current_point.value + previous_point.value\n            output_start_time_unix_nano = previous_point.start_time_unix_nano\n\n        is_monotonic = (\n            previous_point.is_monotonic and current_point.is_monotonic\n        )\n\n        return Sum(\n            start_time_unix_nano=output_start_time_unix_nano,\n            time_unix_nano=current_point.time_unix_nano,\n            value=value,\n            aggregation_temporality=aggregation_temporality,\n            is_monotonic=is_monotonic,\n        )\n\n    return None\n", "path": "opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py"}]}