[mds-stream-processor] Add geo labeling for mds-stream-processor

packages/mds-stream-processor/labelers/geography-labeler.ts

@@ -14,17 +14,30 @@
     limitations under the License.
  */
 
+import db from '@mds-core/mds-db'
 import { UUID, Nullable, Telemetry } from '@mds-core/mds-types'
+import { pointInShape } from '@mds-core/mds-utils'
 import { MessageLabeler } from './types'
 
 export interface GeographyLabel {
-  geography_id: UUID[]
+  geography_ids: UUID[]
 }
 
 export const GeographyLabeler: () => MessageLabeler<
   { telemetry?: Nullable<Telemetry> },
   GeographyLabel
 > = () => async ({ telemetry }) => {
-  // TODO: Add Point-in-polygon checks
-  return { geography_id: telemetry ? [] : [] }
+  const gps = telemetry?.gps
+  if (gps) {
+    const { lat, lng } = gps
+
+    const geographies = await db.readGeographies({ get_published: true })
+
+    const geography_ids = geographies
+      .filter(({ geography_json }) => pointInShape({ lat, lng }, geography_json))
+      .map(({ geography_id }) => geography_id)
+
+    return { geography_ids }
+  }
+  return { geography_ids: [] }
 }

packages/mds-stream-processor/package.json

@@ -28,5 +28,8 @@
     "@mds-core/mds-types": "0.1.23",
     "@mds-core/mds-utils": "0.1.26"
   },
+  "devDependencies": {
+    "sinon": "7.5.0"
+  },
   "license": "Apache-2.0"
 }

packages/mds-stream-processor/tests/labelers/geograpy-labelers.spec.ts

@@ -0,0 +1,41 @@
+import { GeographyLabeler } from '@mds-core/mds-stream-processor/labelers/geography-labeler'
+import * as utils from '@mds-core/mds-utils'
+import db from '@mds-core/mds-db'
+import Sinon from 'sinon'
+import { Geography } from '@mds-core/mds-types'
+import { v4 as uuid } from 'uuid'
+import assert from 'assert'
+
+const mockGeographies: Geography[] = Array.from({ length: 100 }, () => ({
+  geography_id: uuid(),
+  geography_json: {
+    type: 'FeatureCollection',
+    features: []
+  },
+  name: uuid() // Usually intended to be a human readable name, but we just need a random string here.
+}))
+
+const mockTelemetry = {
+  provider_id: uuid(),
+  device_id: uuid(),
+  timestamp: utils.now(),
+  gps: { lat: 34.0522, lng: 118.2437 }
+}
+
+describe('GeographyLabeler tests', async () => {
+  it('Tests all matched geographies are included in list', async () => {
+    Sinon.replace(utils, 'pointInShape', () => true)
+    Sinon.replace(db, 'readGeographies', async () => mockGeographies)
+
+    const { geography_ids } = await GeographyLabeler()({ telemetry: mockTelemetry })
+
+    geography_ids.forEach(geography_id =>
+      assert(mockGeographies.find(geography => geography.geography_id === geography_id))
+    )
+  })
+
+  it('Tests that absent gps in telemetry elicits no matches', async () => {
+    const { geography_ids } = await GeographyLabeler()({})
+    assert(geography_ids.length === [].length)
+  })
+})

packages/mds-stream-processor/tests/labelers/latency-labeler.spec.ts

@@ -0,0 +1,22 @@
+import { LatencyLabeler } from '@mds-core/mds-stream-processor/labelers/latency-labeler'
+import assert from 'assert'
+
+describe('LatencyLabeler Tests', async () => {
+  it('Tests LatencyLabeler detects latency > 0', async () => {
+    const mockTimeChunks = Array.from({ length: 100 }, (_, index) => ({
+      timestamp: 1000 * index,
+      recorded: 1000 * (index + 1)
+    }))
+    const latencyLabels = await Promise.all(mockTimeChunks.map(chunk => LatencyLabeler()(chunk)))
+    latencyLabels.forEach(({ latency_ms }) => assert(latency_ms === 1000))
+  })
+
+  it('Tests LatencyLabeler returns latency == 0', async () => {
+    const mockTimeChunks = Array.from({ length: 100 }, (_, index) => ({
+      timestamp: 1000 * index,
+      recorded: 1000 * index
+    }))
+    const latencyLabels = await Promise.all(mockTimeChunks.map(chunk => LatencyLabeler()(chunk)))
+    latencyLabels.forEach(({ latency_ms }) => assert(latency_ms === 0))
+  })
+})