Retrieve latest datapoint

Get the latest datapoint for one or more time series.

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:

id (int | LatestDatapointQuery | Sequence[int | LatestDatapointQuery] | None) – Id or list of ids.
external_id (str | LatestDatapointQuery | SequenceNotStr[str | LatestDatapointQuery] | None) – External id or list of external ids.
instance_id (NodeId | LatestDatapointQuery | Sequence[NodeId | LatestDatapointQuery] | None) – Instance id or list of instance ids.
before (None | int | str | datetime.datetime) – Get latest datapoint before this time. Not used when passing ‘LatestDatapointQuery’.
target_unit (str | None) – The unit_external_id of the datapoint returned. If the time series does not have a unit_external_id that can be converted to the target_unit, an error will be returned. Cannot be used with target_unit_system.
target_unit_system (str | None) – The unit system of the datapoint returned. Cannot be used with target_unit.
include_status (bool) – Also return the status code, an integer, for each datapoint in the response.
ignore_bad_datapoints (bool) – Prevent datapoints with a bad status code to be returned. Default: True.
treat_uncertain_as_bad (bool) – Treat uncertain status codes as bad. If false, treat uncertain as good. Default: True.
ignore_unknown_ids (bool) – Ignore IDs and external IDs that are not found rather than throw an exception.

Returns:

A LatestDatapoint object containing the latest datapoint (if it exists), or a LatestDatapointList if multiple time series were requested. If ignore_unknown_ids is True, a single time series is requested and it is not found, the function will return None.

Return type:

LatestDatapoint | LatestDatapointList | None

Examples

Getting the latest datapoint in a time series:

>>> from cognite.client import CogniteClient, AsyncCogniteClient
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> res = client.time_series.data.retrieve_latest(id=1)
>>> if res:  # Check if datapoint exists
...     print(res.timestamp, res.value)

You can also use external_id or instance_id; single identifier or list of identifiers:

>>> from cognite.client.data_classes.data_modeling import NodeId
>>> res = client.time_series.data.retrieve_latest(
...     external_id=["foo", "bar"], instance_id=NodeId("my-space", "my-ts-xid")
... )

You can also get the latest datapoint before a specific time:

>>> res = client.time_series.data.retrieve_latest(id=1, before="2d-ago")

You can also get the latest datapoint before a specific time in the future e.g. forecast data:

>>> res = client.time_series.data.retrieve_latest(id=1, before="2d-ahead")

You can also retrieve the datapoint in a different unit or unit system:

>>> res = client.time_series.data.retrieve_latest(id=1, target_unit="temperature:deg_f")
>>> res = client.time_series.data.retrieve_latest(id=1, target_unit_system="Imperial")

You may also pass an instance of LatestDatapointQuery:

>>> from cognite.client.data_classes import LatestDatapointQuery
>>> res = client.time_series.data.retrieve_latest(
...     id=LatestDatapointQuery(id=1, before=60_000)
... )

If you need the latest datapoint for multiple time series, simply give a list of ids. Note that we are using external ids here, but either will work:

>>> res = client.time_series.data.retrieve_latest(external_id=["abc", "def"])
>>> latest_abc = res[0]
>>> latest_def = res[1]

If you for example need to specify a different value of ‘before’ for each time series, you may pass several LatestDatapointQuery objects. These will override any parameter passed directly to the function and also allows for individual customisation of ‘target_unit’, ‘target_unit_system’, ‘include_status’, ‘ignore_bad_datapoints’ and ‘treat_uncertain_as_bad’.

>>> from datetime import datetime, timezone
>>> id_queries = [
...     123,
...     LatestDatapointQuery(id=456, before="1w-ago"),
...     LatestDatapointQuery(id=789, before=datetime(2018, 1, 1, tzinfo=timezone.utc)),
...     LatestDatapointQuery(id=987, target_unit="temperature:deg_f"),
... ]
>>> ext_id_queries = [
...     "foo",
...     LatestDatapointQuery(
...         external_id="abc", before="3h-ago", target_unit_system="Imperial"
...     ),
...     LatestDatapointQuery(external_id="def", include_status=True),
...     LatestDatapointQuery(external_id="ghi", treat_uncertain_as_bad=False),
...     LatestDatapointQuery(
...         external_id="jkl", include_status=True, ignore_bad_datapoints=False
...     ),
... ]
>>> res = client.time_series.data.retrieve_latest(
...     id=id_queries, external_id=ext_id_queries
... )