Custom serialization for dataStream append

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## [2.0.5] - 2025-05-01
+
+### Fixed
+
+- Fixed serialization of unknown datastream `Data`
+
 ## [2.0.4] - 2025-03-10
 
 ### Fixed

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@carp-dk/client",
   "type": "module",
-  "version": "2.0.4",
+  "version": "2.0.5",
   "description": "TypeScript API client for the CARP Web Services (CAWS).",
   "repository": {
     "type": "git",

src/endpoints/dataStreams.ts

@@ -1,6 +1,5 @@
 /* eslint-disable no-underscore-dangle */
 import {
-  DataStreamBatch,
   DataStreamId,
   DataStreamsConfiguration,
   DataStreamServiceRequest,
@@ -57,18 +56,51 @@ class DataStreams extends Endpoint {
     batch,
   }: {
     studyDeploymentId: string;
-    batch: DataStreamBatch;
+    batch: CarpDataStreamBatch;
   }) {
-    const request = new DataStreamServiceRequest.AppendToDataStreams(
-      new UUID(studyDeploymentId),
-      batch,
-    );
-
-    const serializedRequest = serialize({
-      request,
-      serializer: DataStreamServiceRequest.Serializer,
+    const batches = batch.sequences.map((sequence) => {
+      return {
+        dataStream: {
+          studyDeploymentId:
+            sequence.dataStream.studyDeploymentId.stringRepresentation,
+          deviceRoleName: sequence.dataStream.deviceRoleName,
+          dataType: sequence.dataStream.dataType.toString(),
+        },
+        firstSequenceId: sequence.firstSequenceId.toNumber(),
+        measurements: sequence.measurements.toArray().map((measurement) => {
+          return {
+            sensorStartTime: measurement.sensorStartTime.toNumber(),
+            data: {
+              ...Object.fromEntries(
+                // eslint-disable-next-line @typescript-eslint/no-unused-vars
+                Object.entries(measurement.data).filter(([_, value]) => value),
+              ),
+              __type: sequence.dataStream.dataType.toString(),
+            },
+          };
+        }),
+        triggerIds: sequence.triggerIds.toArray(),
+        syncPoint: {
+          synchronizedOn: new Date(
+            sequence.syncPoint.synchronizedOn.toEpochMilliseconds(),
+          ).toISOString(),
+          sensorTimestampAtSyncPoint:
+            sequence.syncPoint.sensorTimestampAtSyncPoint.toNumber(),
+          relativeClockSpeed: sequence.syncPoint.relativeClockSpeed,
+        },
+      };
     });
 
+    const request = {
+      __type:
+        "dk.cachet.carp.data.infrastructure.DataStreamServiceRequest.AppendToDataStreams",
+      apiVersion: "1.1",
+      studyDeploymentId,
+      batch: batches,
+    };
+
+    const serializedRequest = JSON.stringify(request);
+
     await this.actions.post(this.endpoint, serializedRequest);
   }
 

src/test/endpoints/dataStreams.test.ts

@@ -8,7 +8,6 @@ import {
   DefaultSerializer,
   DataStreamsConfiguration,
   NamespacedId,
-  MutableDataStreamBatch,
   MutableDataStreamSequence,
   DataStreamId,
   toLong,
@@ -20,6 +19,7 @@ import {
 import { CarpTestClient } from "@/client";
 import { STUDY_PROTOCOL } from "../consts";
 import { setupTestClient } from "../utils";
+import CarpDataStreamBatch from "@/shared/models/carpDataStreamBatch";
 
 describe("DataStreams", () => {
   let testClient: CarpTestClient;
@@ -105,13 +105,17 @@ describe("DataStreams", () => {
             STUDY_PROTOCOL.primaryDevices[0].roleName,
             namespaceId,
           ),
+          new DataStreamsConfiguration.ExpectedDataStream(
+            STUDY_PROTOCOL.primaryDevices[0].roleName,
+            new NamespacedId("dk.cachet.carp.data", "unknown"),
+          ),
         ],
       }),
     ).resolves.not.toThrow();
   }, 25000);
 
   test("should be able to append to a data stream", async () => {
-    const batch = new MutableDataStreamBatch();
+    const batch = new CarpDataStreamBatch();
     const sequence = new MutableDataStreamSequence(
       new DataStreamId(
         participantGroupStatus.id,
@@ -133,7 +137,43 @@ describe("DataStreams", () => {
       ]),
     );
 
-    batch.appendSequence(sequence);
+    batch.sequences = [sequence];
+
+    await expect(
+      testClient.dataStreams.appendToDataStreams({
+        studyDeploymentId: participantGroupStatus.id.stringRepresentation,
+        batch,
+      }),
+    ).resolves.not.toThrow();
+  });
+
+  test("should be able to append unknown type to a data stream", async () => {
+    const batch = new CarpDataStreamBatch();
+    const sequence = new MutableDataStreamSequence(
+      new DataStreamId(
+        participantGroupStatus.id,
+        STUDY_PROTOCOL.primaryDevices[0].roleName,
+        new NamespacedId("dk.cachet.carp.data", "unknown"),
+      ),
+      toLong(2),
+      toList([1]),
+      SyncPoint.Companion.UnixEpoch,
+    );
+    sequence.appendMeasurementsList(
+      toList([
+        new Measurement(
+          toLong(1),
+          null,
+          new NamespacedId("dk.cachet.carp.data", "unknown"),
+          {
+            value: 1,
+            unit: "unknown",
+          } as any,
+        ),
+      ]),
+    );
+
+    batch.sequences = [sequence];
 
     await expect(
       testClient.dataStreams.appendToDataStreams({
@@ -144,7 +184,7 @@ describe("DataStreams", () => {
   });
 
   test("should be able to get data streams", async () => {
-    const batch = new MutableDataStreamBatch();
+    const batch = new CarpDataStreamBatch();
     const sequence = new MutableDataStreamSequence(
       new DataStreamId(
         participantGroupStatus.id,
@@ -166,7 +206,32 @@ describe("DataStreams", () => {
       ]),
     );
 
-    batch.appendSequence(sequence);
+    const sequence2 = new MutableDataStreamSequence(
+      new DataStreamId(
+        participantGroupStatus.id,
+        STUDY_PROTOCOL.primaryDevices[0].roleName,
+        new NamespacedId("dk.cachet.carp.data", "unknown"),
+      ),
+      toLong(0),
+      toList([1]),
+      SyncPoint.Companion.UnixEpoch,
+    );
+
+    sequence2.appendMeasurementsList(
+      toList([
+        new Measurement(
+          toLong(1),
+          null,
+          new NamespacedId("dk.cachet.carp.data", "unknown"),
+          {
+            value: 1,
+            unit: "unknown",
+          } as any,
+        ),
+      ]),
+    );
+
+    batch.sequences = [sequence, sequence2];
 
     await expect(
       testClient.dataStreams.appendToDataStreams({
@@ -175,7 +240,7 @@ describe("DataStreams", () => {
       }),
     ).resolves.not.toThrow();
 
-    const response = await testClient.dataStreams.getDataStream({
+    let response = await testClient.dataStreams.getDataStream({
       dataStream: new DataStreamId(
         participantGroupStatus.id,
         STUDY_PROTOCOL.primaryDevices[0].roleName,
@@ -200,6 +265,32 @@ describe("DataStreams", () => {
       .forEach((point) => {
         expect(point).toBeInstanceOf(Measurement);
       });
+
+    response = await testClient.dataStreams.getDataStream({
+      dataStream: new DataStreamId(
+        participantGroupStatus.id,
+        STUDY_PROTOCOL.primaryDevices[0].roleName,
+        new NamespacedId("dk.cachet.carp.data", "unknown"),
+      ),
+      fromSequenceId: 0,
+    });
+
+    expect(response.isEmpty()).toBe(false);
+    expect(response.sequences.length).to.be.at.least(1);
+    expect(response.sequences[0].measurements.toArray().length).to.be.at.least(
+      1,
+    );
+    response
+      .getDataStreamPoints(
+        new DataStreamId(
+          participantGroupStatus.id,
+          STUDY_PROTOCOL.primaryDevices[0].roleName,
+          new NamespacedId("dk.cachet.carp.data", "unknown"),
+        ),
+      )
+      .forEach((point) => {
+        expect(point).toBeInstanceOf(Measurement);
+      });
   });
 
   afterAll(async () => {