Support inserting datapoints with status codes (#1707)

CHANGELOG.md

@@ -17,6 +17,10 @@ Changes are grouped as follows
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+## [7.35.0] - 2024-04-16
+### Added
+- Datapoints insert methods `insert` and `insert_multiple` now support ingesting (optional) status codes.
+
 ## [7.34.0] - 2024-04-11
 ### Added
 - Datapoints method `retrieve_latest` now supports status codes.

cognite/client/_api/datapoint_tasks.py

@@ -33,6 +33,7 @@
 from google.protobuf.internal.containers import RepeatedCompositeFieldContainer
 from typing_extensions import NotRequired, TypeAlias
 
+from cognite.client._constants import NUMPY_IS_AVAILABLE
 from cognite.client._proto.data_point_list_response_pb2 import DataPointListItem
 from cognite.client._proto.data_points_pb2 import (
     AggregateDatapoint,
@@ -43,7 +44,6 @@
 from cognite.client.data_classes.datapoints import (
     _AGGREGATES_IN_BETA,
     _INT_AGGREGATES,
-    NUMPY_IS_AVAILABLE,
     Aggregate,
     Datapoints,
     DatapointsArray,

cognite/client/_constants.py

@@ -10,3 +10,11 @@
 # Max JavaScript-safe integer 2^53 - 1
 MAX_VALID_INTERNAL_ID = 9007199254740991
 DATA_MODELING_DEFAULT_LIMIT_READ = 10
+
+try:
+    import numpy as np  # noqa F401
+
+    NUMPY_IS_AVAILABLE = True
+
+except ImportError:  # pragma no cover
+    NUMPY_IS_AVAILABLE = False

cognite/client/_version.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
-__version__ = "7.34.0"
+__version__ = "7.35.0"
 __api_subversion__ = "20230101"

cognite/client/data_classes/__init__.py

@@ -46,6 +46,7 @@
     DatapointsList,
     DatapointsQuery,
     LatestDatapointQuery,
+    StatusCode,
 )
 from cognite.client.data_classes.datapoints_subscriptions import (
     DatapointSubscription,
@@ -505,6 +506,7 @@
     "DatapointsArrayList",
     "DatapointsQuery",
     "LatestDatapointQuery",
+    "StatusCode",
     "Function",
     "FunctionWrite",
     "FunctionFilter",

cognite/client/data_classes/datapoints.py

@@ -6,6 +6,7 @@
 from collections import defaultdict
 from dataclasses import InitVar, dataclass, fields
 from datetime import datetime
+from enum import IntEnum
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -16,6 +17,7 @@
     overload,
 )
 
+from cognite.client._constants import NUMPY_IS_AVAILABLE
 from cognite.client.data_classes._base import CogniteResource, CogniteResourceList
 from cognite.client.utils import _json
 from cognite.client.utils._auxiliary import find_duplicates
@@ -34,6 +36,13 @@
 from cognite.client.utils._time import convert_and_isoformat_time_attrs
 from cognite.client.utils.useful_types import SequenceNotStr
 
+
+class StatusCode(IntEnum):
+    Good = 0x0
+    Uncertain = 0x40000000  # aka 1 << 30 aka 1073741824
+    Bad = 0x80000000  # aka 1 << 31 aka 2147483648
+
+
 Aggregate = Literal[
     "average",
     "continuous_variance",
@@ -81,14 +90,9 @@
 )
 ALL_SORTED_DP_AGGS = sorted(typing.get_args(Aggregate))
 
-try:
+if NUMPY_IS_AVAILABLE:
     import numpy as np
 
-    NUMPY_IS_AVAILABLE = True
-
-except ImportError:  # pragma no cover
-    NUMPY_IS_AVAILABLE = False
-
 if TYPE_CHECKING:
     import numpy.typing as npt
     import pandas
@@ -448,12 +452,14 @@ def __getitem__(self, item: int | slice) -> Datapoint | DatapointsArray:
             return self._slice(item)
         attrs, arrays = self._data_fields()
         timestamp = arrays[0][item].item() // 1_000_000
-        data = {attr: numpy_dtype_fix(arr[item]) for attr, arr in zip(attrs[1:], arrays[1:])}
+        data: dict[str, float | str | None] = {
+            attr: numpy_dtype_fix(arr[item]) for attr, arr in zip(attrs[1:], arrays[1:])
+        }
 
         if self.status_code is not None:
             data.update(status_code=self.status_code[item], status_symbol=self.status_symbol[item])  # type: ignore [index]
         if self.null_timestamps and timestamp in self.null_timestamps:
-            data["value"] = None  # type: ignore [assignment]
+            data["value"] = None
         return Datapoint(timestamp=timestamp, **data)  # type: ignore [arg-type]
 
     def _slice(self, part: slice) -> DatapointsArray:
@@ -487,11 +493,11 @@ def __iter__(self) -> Iterator[Datapoint]:
         # Let's not create a single Datapoint more than we have too:
         for i, row in enumerate(zip(*arrays)):
             timestamp = row[0].item() // 1_000_000
-            data = dict(zip(attrs[1:], map(numpy_dtype_fix, row[1:])))
+            data: dict[str, float | str | None] = dict(zip(attrs[1:], map(numpy_dtype_fix, row[1:])))
             if self.status_code is not None:
                 data.update(status_code=self.status_code[i], status_symbol=self.status_symbol[i])  # type: ignore [index]
             if self.null_timestamps and timestamp in self.null_timestamps:
-                data["value"] = None  # type: ignore [assignment]
+                data["value"] = None
 
             yield Datapoint(timestamp=timestamp, **data)  # type: ignore [arg-type]
 
@@ -539,8 +545,11 @@ def dump(self, camel_case: bool = True, convert_timestamps: bool = False) -> dic
 
             for dp, code, symbol in zip(datapoints, map(numpy_dtype_fix, self.status_code), self.status_symbol):
                 dp["status"] = {"code": code, "symbol": symbol}  # type: ignore [assignment]
-                # When we're dealing with status codes, NaN might be either one of [<missing>, nan]:
-                if dp["timestamp"] in (self.null_timestamps or ()):  # ...luckily, we know :3
+
+        # When we're dealing with datapoints with bad status codes, NaN might be either one of [<missing>, nan]:
+        if self.null_timestamps:
+            for dp in datapoints:
+                if dp["timestamp"] in self.null_timestamps:  # ...luckily, we know :3
                     dp["value"] = None  # type: ignore [assignment]
         dumped["datapoints"] = datapoints
 

cognite/client/utils/_json.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import json
+import math
 import numbers
 from decimal import Decimal
+from types import MappingProxyType
 from typing import Any
 
-__all__ = ["dumps", "loads"]
+__all__ = ["dumps", "loads", "convert_to_float", "convert_nonfinite_float_to_str"]
 
 
 def _default_json_encoder(obj: Any) -> Any:
@@ -41,3 +43,31 @@ def dumps(
 
 
 loads = json.loads
+
+# As opposed to protobuf, datapoints in JSON returns out-of-range float values as strings. This means
+# we're forced to do a lookup for every single datapoint we try to insert (and read in retrieve_latest
+# when not ignoring bad)... so we allow some ugly optimizations in the translation code:
+_FLOAT_API_MAPPING = MappingProxyType({"Infinity": math.inf, "-Infinity": -math.inf, "NaN": math.nan})
+_FLOAT_API_MAPPING_REVERSE = MappingProxyType({math.inf: "Infinity", -math.inf: "-Infinity", math.nan: "NaN"})
+
+
+def convert_to_float(value: float | str | None) -> float | None:
+    if value.__class__ is str:  # like this abomination; faster than float(value)
+        return _FLOAT_API_MAPPING[value]  # type: ignore [index]
+    return value  # type: ignore [return-value]
+
+
+def convert_nonfinite_float_to_str(value: float | str | None) -> float | str | None:
+    # We accept str because when a user is trying to insert datapoints - we have no idea if the
+    # time series to insert into is string or numeric
+    try:
+        return value if math.isfinite(value) else _FLOAT_API_MAPPING_REVERSE[value]  # type: ignore [arg-type, index]
+    except TypeError:
+        if value.__class__ is str or value is None:
+            return value
+        raise
+    except KeyError:
+        # Depending on numpy and python version, dict lookup may fail for NaN.. thanks IEEE :wink:
+        if math.isnan(value):  # type: ignore [arg-type]
+            return "NaN"
+        raise

cognite/client/utils/useful_types.py

@@ -21,7 +21,7 @@ def __len__(self) -> int: ...
 
     def __iter__(self) -> Iterator[_T_co]: ...
 
-    def index(self, value: Any, /, start: int = 0, stop: int = ...) -> int: ...
+    def index(self, value: Any, start: int = 0, stop: int = ..., /) -> int: ...
 
     def count(self, value: Any, /) -> int: ...
 

docs/source/time_series.rst

@@ -79,8 +79,8 @@ Calculate the result of a function on time series
 
 
 
-Data points
------------
+Datapoints
+----------
 
 Retrieve datapoints
 ^^^^^^^^^^^^^^^^^^^
@@ -102,46 +102,46 @@ Retrieve latest datapoint
 ^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints.DatapointsAPI.retrieve_latest
 
-Insert data points
-^^^^^^^^^^^^^^^^^^
+Insert datapoints
+^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints.DatapointsAPI.insert
 
-Insert data points into multiple time series
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Insert datapoints into multiple time series
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints.DatapointsAPI.insert_multiple
 
 Insert pandas dataframe
 ^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints.DatapointsAPI.insert_dataframe
 
-Delete a range of data points
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Delete a range of datapoints
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints.DatapointsAPI.delete_range
 
-Delete ranges of data points
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Delete ranges of datapoints
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints.DatapointsAPI.delete_ranges
 
 
-Data Points Data classes
-^^^^^^^^^^^^^^^^^^^^^^^^
+Datapoints Data classes
+^^^^^^^^^^^^^^^^^^^^^^^
 .. automodule:: cognite.client.data_classes.datapoints
     :members:
     :show-inheritance:
 
-Data Point Subscriptions
+Datapoint Subscriptions
 ---------------------------
 
-Create data point subscriptions
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Create datapoint subscriptions
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints_subscriptions.DatapointsSubscriptionAPI.create
 
-Retrieve a data point subscription by id(s)
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Retrieve a datapoint subscription by id(s)
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints_subscriptions.DatapointsSubscriptionAPI.retrieve
 
-List data point subscriptions
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+List datapoint subscriptions
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints_subscriptions.DatapointsSubscriptionAPI.list
 
 List member time series of subscription
@@ -152,15 +152,15 @@ Iterate over subscriptions data
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints_subscriptions.DatapointsSubscriptionAPI.iterate_data
 
-Update data point subscription
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Update datapoint subscription
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints_subscriptions.DatapointsSubscriptionAPI.update
 
-Delete data point subscription
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Delete datapoint subscription
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automethod:: cognite.client._api.datapoints_subscriptions.DatapointsSubscriptionAPI.delete
 
-Data Point Subscription classes
+Datapoint Subscription classes
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 .. automodule:: cognite.client.data_classes.datapoints_subscriptions
     :members:

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "cognite-sdk"
 
-version = "7.34.0"
+version = "7.35.0"
 description = "Cognite Python SDK"
 readme = "README.md"
 documentation = "https://cognite-sdk-python.readthedocs-hosted.com"