feat(measures): add lastMeasuredAt on DigitalTwin (#314)

kuzzleio · Aug 21, 2023 · 05b8a53 · 05b8a53
1 parent 0c3cab5
commit 05b8a53
Show file tree

Hide file tree

Showing 11 changed files with 145 additions and 27 deletions.
diff --git a/lib/modules/asset/collections/assetsMappings.ts b/lib/modules/asset/collections/assetsMappings.ts
@@ -38,6 +38,8 @@ export const assetsMappings: CollectionMappings = {
       },
     },
 
+    lastMeasuredAt: { type: "date" },
+
     linkedDevices: {
       properties: {
         _id: { type: "keyword" },

diff --git a/lib/modules/decoder/PayloadService.ts b/lib/modules/decoder/PayloadService.ts
@@ -227,6 +227,7 @@ export class PayloadService {
       const body: DeviceContent = {
         assetId: null,
         engineId: null,
+        lastMeasuredAt: 0,
         measures: {},
         metadata: {},
         model: deviceModel,

diff --git a/lib/modules/device/DeviceService.ts b/lib/modules/device/DeviceService.ts
@@ -88,6 +88,7 @@ export class DeviceService {
       _source: {
         assetId: null,
         engineId: null,
+        lastMeasuredAt: 0,
         measures: {},
         metadata,
         model,

diff --git a/lib/modules/device/collections/deviceMappings.ts b/lib/modules/device/collections/deviceMappings.ts
@@ -32,5 +32,6 @@ export const devicesMappings = {
         // populated with measure models
       },
     },
+    lastMeasuredAt: { type: "date" },
   },
 };
diff --git a/lib/modules/measure/MeasureService.ts b/lib/modules/measure/MeasureService.ts
@@ -278,6 +278,8 @@ export class MeasureService {
       device._source.measures = {};
     }
 
+    let lastMeasuredAt = 0;
+
     for (const measurement of measurements) {
       if (measurement.origin.type === "computed") {
         continue;
@@ -293,6 +295,10 @@ export class MeasureService {
         continue;
       }
 
+      if (measurement.measuredAt > lastMeasuredAt) {
+        lastMeasuredAt = measurement.measuredAt;
+      }
+
       device._source.measures[measureName] = {
         measuredAt: measurement.measuredAt,
         name: measureName,
@@ -302,6 +308,8 @@ export class MeasureService {
         values: measurement.values,
       };
     }
+
+    device._source.lastMeasuredAt = lastMeasuredAt;
   }
 
   // @todo there shouldn't be any logic related to asset historization here, but no other choices for now. It needs to be re-architected
@@ -321,6 +329,8 @@ export class MeasureService {
       asset._source.measures = {};
     }
 
+    let lastMeasuredAt = 0;
+
     for (const measurement of measurements) {
       if (measurement.origin.type === "computed") {
         continue;
@@ -345,6 +355,10 @@ export class MeasureService {
         continue;
       }
 
+      if (measurement.measuredAt > lastMeasuredAt) {
+        lastMeasuredAt = measurement.measuredAt;
+      }
+
       asset._source.measures[measureName] = {
         measuredAt: measurement.measuredAt,
         name: measureName,
@@ -360,6 +374,8 @@ export class MeasureService {
       );
     }
 
+    asset._source.lastMeasuredAt = lastMeasuredAt;
+
     return assetStates;
   }
 

diff --git a/lib/modules/shared/types/DigitalTwinContent.ts b/lib/modules/shared/types/DigitalTwinContent.ts
@@ -16,4 +16,6 @@ export interface DigitalTwinContent<
   measures: {
     [Property in keyof TMeasures]: EmbeddedMeasure<TMeasures[Property]> | null;
   };
+
+  lastMeasuredAt: number;
 }
diff --git a/tests/application/decoders/DummyTempDecoder.ts b/tests/application/decoders/DummyTempDecoder.ts
@@ -8,6 +8,7 @@ import {
   BatteryMeasurement,
 } from "../../../index";
 import { AccelerationMeasurement } from "../measures/AccelerationMeasure";
+import { isMeasureDated } from "../../helpers/payloads";
 
 export class DummyTempDecoder extends Decoder {
   public measures = [
@@ -67,28 +68,38 @@ export class DummyTempDecoder extends Decoder {
       payload.deviceEUI,
       "temperature",
       {
-        measuredAt: payload.measuredAt || Date.now(),
+        measuredAt: isMeasureDated(payload.temperature)
+          ? payload.temperature.measuredAt
+          : payload.measuredAt ?? Date.now(),
         type: "temperature",
         values: {
-          temperature: payload.temperature,
+          temperature: isMeasureDated(payload.temperature)
+            ? payload.temperature.value
+            : payload.temperature,
         },
       }
     );
 
     if (payload.acceleration !== undefined) {
+      const acceleration = isMeasureDated(payload.acceleration)
+        ? payload.acceleration.value
+        : payload.acceleration;
+
       decodedPayload.addMeasurement<AccelerationMeasurement>(
         payload.deviceEUI,
         "accelerationSensor",
         {
-          measuredAt: payload.measuredAt || Date.now(),
+          measuredAt: isMeasureDated(payload.acceleration)
+            ? payload.acceleration.measuredAt
+            : payload.measuredAt ?? Date.now(),
           type: "acceleration",
           values: {
             acceleration: {
-              x: payload.acceleration.x,
-              y: payload.acceleration.y,
-              z: payload.acceleration.z,
+              x: acceleration.x,
+              y: acceleration.y,
+              z: acceleration.z,
             },
-            accuracy: payload.acceleration.accuracy,
+            accuracy: acceleration.accuracy,
           },
         }
       );
@@ -98,10 +109,14 @@ export class DummyTempDecoder extends Decoder {
       payload.deviceEUI,
       "battery",
       {
-        measuredAt: payload.measuredAt || Date.now(),
+        measuredAt: isMeasureDated(payload.battery)
+          ? payload.battery.measuredAt
+          : payload.measuredAt ?? Date.now(),
         type: "battery",
         values: {
-          battery: payload.battery || 42,
+          battery: isMeasureDated(payload.battery)
+            ? payload.battery.value
+            : payload.battery,
         },
       }
     );

diff --git a/tests/application/decoders/DummyTempPositionDecoder.ts b/tests/application/decoders/DummyTempPositionDecoder.ts
@@ -8,6 +8,7 @@ import {
   TemperatureMeasurement,
   DecodedPayload,
 } from "../../../index";
+import { isMeasureDated } from "../../helpers/payloads";
 
 export class DummyTempPositionDecoder extends Decoder {
   public measures = [
@@ -32,36 +33,52 @@ export class DummyTempPositionDecoder extends Decoder {
       payload.deviceEUI,
       "temperature",
       {
-        measuredAt: Date.now(),
+        measuredAt: isMeasureDated(payload.temperature)
+          ? payload.temperature.measuredAt
+          : payload.measuredAt ?? Date.now(),
         type: "temperature",
-        values: { temperature: payload.temperature },
+        values: {
+          temperature: isMeasureDated(payload.temperature)
+            ? payload.temperature.value
+            : payload.temperature,
+        },
       }
     );
 
+    const location = isMeasureDated(payload.location)
+      ? payload.location.value
+      : payload.location;
     decodedPayload.addMeasurement<PositionMeasurement>(
       payload.deviceEUI,
       "position",
       {
-        measuredAt: Date.now(),
+        measuredAt: isMeasureDated(payload.location)
+          ? payload.location.measuredAt
+          : payload.measuredAt ?? Date.now(),
         type: "position",
         values: {
           position: {
-            lat: payload.location.lat,
-            lon: payload.location.lon,
+            lat: location.lat,
+            lon: location.lon,
           },
-          accuracy: payload.location.accuracy,
+          accuracy: location.accuracy,
         },
       }
     );
 
+    const battery = isMeasureDated(payload.battery)
+      ? payload.battery.value
+      : payload.battery;
     decodedPayload.addMeasurement<BatteryMeasurement>(
       payload.deviceEUI,
       "battery",
       {
-        measuredAt: Date.now(),
+        measuredAt: isMeasureDated(payload.battery)
+          ? payload.battery.measuredAt
+          : payload.measuredAt ?? Date.now(),
         type: "battery",
         values: {
-          battery: payload.battery * 100,
+          battery: battery * 100,
         },
       }
     );

diff --git a/tests/helpers/payloads.ts b/tests/helpers/payloads.ts
@@ -1,10 +1,38 @@
 import { JSONObject, Kuzzle } from "kuzzle-sdk";
 
+type MeasureDated<T> = { value: T; measuredAt: number };
+export type MeasureValue<T> = T | MeasureDated<T>;
+
+export function isMeasureDated<T>(
+  measure: MeasureValue<T>
+): measure is MeasureDated<T> {
+  if (measure === undefined || measure === null) {
+    return false;
+  }
+
+  const test = measure as MeasureDated<T>;
+  return test.value !== undefined && test.measuredAt !== undefined;
+}
+
+export interface Location {
+  lat: number;
+  lon: number;
+  accuracy?: number;
+}
+
+export interface Acceleration {
+  x: number;
+  y: number;
+  z: number;
+  accuracy: number;
+}
+
 export type DummyTempSimplePayload = {
   deviceEUI: string;
-  temperature: number;
+  temperature: MeasureValue<number>;
   measuredAt?: number;
-  battery?: number;
+  battery?: MeasureValue<number>;
+  acceleration?: MeasureValue<Acceleration>;
   metadata?: JSONObject;
 };
 
@@ -15,11 +43,7 @@ export type DummyTempPayload =
     };
 
 export type DummyTempPositionPayload = DummyTempPayload & {
-  location: {
-    lat: number;
-    lon: number;
-    accuracy?: number;
-  };
+  location: MeasureValue<Location>;
 };
 
 export async function sendDummyTempPayloads(

diff --git a/tests/scenario/modules/ingestion-pipeline/pipeline-persist-before.test.ts b/tests/scenario/modules/ingestion-pipeline/pipeline-persist-before.test.ts
@@ -1,5 +1,3 @@
-import { ContainerAssetContent } from "../../../application/assets/Container";
-
 import { sendDummyTempPayloads, setupHooks } from "../../../helpers";
 
 jest.setTimeout(10000);

diff --git a/tests/scenario/modules/ingestion-pipeline/pipeline-process-before.test.ts b/tests/scenario/modules/ingestion-pipeline/pipeline-process-before.test.ts
@@ -1,6 +1,11 @@
+import { DeviceContent } from "lib/modules/device";
 import { ContainerAssetContent } from "../../../application/assets/Container";
 
-import { sendDummyTempPayloads, setupHooks } from "../../../helpers";
+import {
+  sendDummyTempPayloads,
+  sendDummyTempPositionPayloads,
+  setupHooks,
+} from "../../../helpers";
 
 jest.setTimeout(10000);
 
@@ -37,4 +42,40 @@ describe("Ingestion Pipeline: process before", () => {
       },
     });
   });
+
+  it("should update lastReceive for new measures", async () => {
+    const now = Date.now();
+    await sendDummyTempPositionPayloads(sdk, [
+      {
+        deviceEUI: "linked2",
+        temperature: {
+          value: 21,
+          measuredAt: 1680096420000, // 13:27:00 UTC
+        },
+        location: {
+          value: {
+            lat: 21,
+            lon: 21,
+          },
+          measuredAt: 1680096300000, // 13:25:00 UTC
+        },
+      },
+    ]);
+
+    const device = await sdk.document.get<DeviceContent>(
+      "engine-ayse",
+      "devices",
+      "DummyTempPosition-linked2"
+    );
+
+    expect(device._source.lastMeasuredAt).toBeGreaterThanOrEqual(now);
+
+    const asset = await sdk.document.get<ContainerAssetContent>(
+      "engine-ayse",
+      "assets",
+      "Container-linked2"
+    );
+
+    expect(asset._source.lastMeasuredAt).toBe(1680096420000);
+  });
 });