Insert datapoints

Insert datapoints into a time series.

Timestamps can be represented as milliseconds since epoch or datetime objects. Note that naive datetimes are interpreted to be in the local timezone (not UTC), adhering to Python conventions for datetime handling.

Time series support status codes like Good, Uncertain and Bad. You can read more in the Cognite Data Fusion developer documentation on status codes.

Parameters:

datapoints (Datapoints | DatapointsArray | Sequence[dict[str, int | float | str | datetime.datetime]] | Sequence[tuple[int | float | datetime.datetime, int | float | str] | tuple[int | float | datetime.datetime, int | float | str, int]]) – The datapoints you wish to insert. Can either be a list of tuples, a list of dictionaries, a Datapoints object or a DatapointsArray object. See examples below.
id (int | None) – Id of time series to insert datapoints into.
external_id (str | None) – External id of time series to insert datapoint into.
instance_id (NodeId | None) – Instance ID of time series to insert datapoints into.

Note

All datapoints inserted without a status code (or symbol) is assumed to be good (code 0). To mark a value, pass either the status code (int) or status symbol (str). Only one of code and symbol is required. If both are given, they must match or an API error will be raised.

Datapoints marked bad can take on any of the following values: None (missing), NaN, and +/- Infinity. It is also not restricted by the normal numeric range [-1e100, 1e100] (i.e. can be any valid float64).

Examples

Your datapoints can be a list of tuples where the first element is the timestamp and the second element is the value. The third element is optional and may contain the status code for the datapoint. To pass by symbol, a dictionary must be used.

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import StatusCode
>>> from datetime import datetime, timezone
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> datapoints = [
...     (datetime(2018, 1, 1, tzinfo=timezone.utc), 1000),
...     (datetime(2018, 1, 2, tzinfo=timezone.utc), 2000, StatusCode.Good),
...     (datetime(2018, 1, 3, tzinfo=timezone.utc), 3000, StatusCode.Uncertain),
...     (datetime(2018, 1, 4, tzinfo=timezone.utc), None, StatusCode.Bad),
... ]
>>> client.time_series.data.insert(datapoints, id=1)

The timestamp can be given by datetime as above, or in milliseconds since epoch. Status codes can also be passed as normal integers; this is necessary if a subcategory or modifier flag is needed, e.g. 3145728: ‘GoodClamped’:

>>> from cognite.client.data_classes.data_modeling import NodeId
>>> datapoints = [
...     (150000000000, 1000),
...     (160000000000, 2000, 3145728),
...     (170000000000, 2000, 2147483648),  # Same as StatusCode.Bad
... ]
>>> client.time_series.data.insert(
...     datapoints, instance_id=NodeId("my-space", "my-ts-xid")
... )

Or they can be a list of dictionaries:

>>> import math
>>> datapoints = [
...     {"timestamp": 150000000000, "value": 1000},
...     {"timestamp": 160000000000, "value": 2000},
...     {"timestamp": 170000000000, "value": 3000, "status": {"code": 0}},
...     {"timestamp": 180000000000, "value": 4000, "status": {"symbol": "Uncertain"}},
...     {
...         "timestamp": 190000000000,
...         "value": math.nan,
...         "status": {"code": StatusCode.Bad, "symbol": "Bad"},
...     },
... ]
>>> client.time_series.data.insert(datapoints, external_id="abcd")

Or they can be a Datapoints or DatapointsArray object (with raw datapoints only). Note that the id or external_id set on these objects are not inspected/used (as they belong to the “from-time-series”, and not the “to-time-series”), and so you must explicitly pass the identifier of the time series you want to insert into, which in this example is external_id=”foo”.

If the Datapoints or DatapointsArray are fetched with status codes, these will be automatically used in the insert:

>>> data = client.time_series.data.retrieve(
...     external_id="abc",
...     start="1w-ago",
...     end="now",
...     include_status=True,
...     ignore_bad_datapoints=False,
... )
>>> client.time_series.data.insert(data, external_id="foo")