py-test-proto-json.py

from timeit import default_timer as timer

import cognite.client.proto.data_point_list_response_pb2 as ProtobufDataPointList

from cognite.client.data_classes import Datapoints

payload = dict(
    start=32760000000-1,
    end=1638000000000,
    ignoreUnknownIds=False,
    limit=100_000,
    includeOutsidePoints=False,
    items=[
        {
            "externalId": "ts-test-#04-ten-mill-dps-1/1",
        },
    ]
)
client = forge_client
res = client.datapoints._post(client.datapoints._RESOURCE_PATH + "/list", json=payload)
asdf = res
print(f"Query (JSON): {round(res.elapsed.total_seconds()*1000, 6)} ms")
t0 = timer()
dps = res.json()["items"]
t1 = timer()
print(f"Parse JSON: {round((t1-t0)*1000, 6)} ms")

t0 = timer()
dps1 = Datapoints(
    timestamp=[dp["timestamp"] for dp in dps[0]["datapoints"]],
    value=[dp["value"] for dp in dps[0]["datapoints"]],
)
t1 = timer()
print(f"Convert to Datapoints: {round((t1-t0)*1000, 6)} ms")
# print("n dps1:", len(dps1[0]["datapoints"]))
# print(dps1.to_pandas())

headers = {"accept": "application/protobuf"}
url_path = client.datapoints._RESOURCE_PATH + "/list"

res = client.datapoints._do_request(
    "POST", url_path, json=payload, headers=headers, params=None, timeout=client._config.timeout
)
print(f"Query (PROTO): {round(res.elapsed.total_seconds()*1000, 6)} ms")
t0 = timer()
dps2 = ProtobufDataPointList.DataPointListResponse()
dps2.ParseFromString(res.content)
# dps2.items[0].numericDatapoints.datapoints[0].timestamp
t1 = timer()
print(f"Parse PROTOBUF: {round((t1-t0)*1000, 6)} ms")

t0 = timer()
dps2 = Datapoints(
    timestamp=[dp.timestamp for dp in dps2.items[0].numericDatapoints.datapoints],
    value=[dp.value for dp in dps2.items[0].numericDatapoints.datapoints],
)
t1 = timer()
print(f"Convert to Datapoints: {round((t1-t0)*1000, 6)} ms")

assert len(dps1) == 100_000
assert len(dps2) == 100_000
assert dps1 == dps2