Converters for the New Data Types: Floors, SedentaryPeriod, and ActivityLevels

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthActivityLevelAvroConverter.kt

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2026 King's College London
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.radarbase.googlehealth.converter
+
+import com.fasterxml.jackson.databind.JsonNode
+import org.apache.avro.specific.SpecificRecord
+import org.radarbase.googlehealth.user.User
+import org.radarbase.googlehealth.util.googleHealthActivityLevel
+import org.radarcns.push.googlehealth.GoogleHealthActivityLevelType
+
+/**
+ * Converts `activity-level` data points: the activity level the user sustained over an interval,
+ * one minute long in practice, reported for every interval of the day. The sedentary intervals are
+ * also reported grouped into longer periods, see [GoogleHealthSedentaryPeriodAvroConverter].
+ */
+class GoogleHealthActivityLevelAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
+    override fun convertDataPoint(
+        point: JsonNode,
+        user: User,
+    ): List<Pair<SpecificRecord, SpecificRecord>> {
+        val data = point["activityLevel"] ?: return emptyList()
+        val (start, end) = parseInterval(data) ?: return emptyList()
+        val record = googleHealthActivityLevel {
+            time = epochSeconds(start)
+            timeReceived = nowEpochSeconds()
+            timeInterval = (end.epochSecond - start.epochSecond).toInt().coerceAtLeast(0)
+            level = mapLevel(data["activityLevelType"]?.asText())
+        }
+        return listOf(user.observationKey to record)
+    }
+
+    /**
+     * Google's `ACTIVITY_LEVEL_TYPE_UNSPECIFIED` is kept as its own symbol, it means Google itself
+     * did not classify the interval. Any other symbol, including ones Google adds later, is
+     * `UNKNOWN`.
+     */
+    private fun mapLevel(text: String?): GoogleHealthActivityLevelType = when (text) {
+        "SEDENTARY" -> GoogleHealthActivityLevelType.SEDENTARY
+        "LIGHTLY_ACTIVE" -> GoogleHealthActivityLevelType.LIGHTLY_ACTIVE
+        "MODERATELY_ACTIVE" -> GoogleHealthActivityLevelType.MODERATELY_ACTIVE
+        "VERY_ACTIVE" -> GoogleHealthActivityLevelType.VERY_ACTIVE
+        "ACTIVITY_LEVEL_TYPE_UNSPECIFIED" ->
+            GoogleHealthActivityLevelType.ACTIVITY_LEVEL_TYPE_UNSPECIFIED
+        else -> GoogleHealthActivityLevelType.UNKNOWN
+    }
+}

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthAvroConverter.kt

@@ -21,6 +21,7 @@ import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
+import java.io.IOException
 import java.time.Instant
 import java.time.LocalDate
 import java.time.LocalDateTime
@@ -31,6 +32,7 @@ abstract class GoogleHealthAvroConverter(override val topic: String) : AvroConve
 
     protected val logger: Logger = LoggerFactory.getLogger(javaClass)
 
+    @Throws(IOException::class)
     abstract fun convertDataPoint(
         point: JsonNode,
         user: User,

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/DailyRestingHeartRateGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthDailyRestingHeartRateAvroConverter.kt

@@ -21,7 +21,7 @@ import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthDailyRestingHeartRate
 
-class DailyRestingHeartRateGoogleHealthAvroConverter(topic: String) :
+class GoogleHealthDailyRestingHeartRateAvroConverter(topic: String) :
     GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/DailySleepTemperatureDerivationsGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthDailySleepTemperatureDerivationsAvroConverter.kt

@@ -21,15 +21,19 @@ import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthDailySleepTemperatureDerivations
 
-class DailySleepTemperatureDerivationsGoogleHealthAvroConverter(topic: String) :
+class GoogleHealthDailySleepTemperatureDerivationsAvroConverter(topic: String) :
     GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,
         user: User,
     ): List<Pair<SpecificRecord, SpecificRecord>> {
         val data = point["dailySleepTemperatureDerivations"] ?: return emptyList()
-        val nightly = data["nightlyTemperatureCelsius"]?.takeIf { it.isNumber }?.floatValue() ?: return emptyList()
-        val baseline = data["baselineTemperatureCelsius"]?.takeIf { it.isNumber }?.floatValue() ?: return emptyList()
+        val nightly = data["nightlyTemperatureCelsius"]?.takeIf {
+            it.isNumber
+        }?.floatValue() ?: return emptyList()
+        val baseline = data["baselineTemperatureCelsius"]?.takeIf {
+            it.isNumber
+        }?.floatValue() ?: return emptyList()
         // `date` is the civil date (in the user's timezone) the derivation is for — emit it
         // directly as a yyyy-MM-dd string, like DailyRestingHeartRate, rather than as a
         // UTC-midnight instant that could shift to the wrong local day downstream.

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/ElectrocardiogramGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthElectrocardiogramAvroConverter.kt

@@ -20,7 +20,10 @@ import com.fasterxml.jackson.databind.JsonNode
 import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthElectrocardiogram
+import java.io.IOException
 import java.time.Instant
+import java.util.stream.Collectors
+import java.util.stream.IntStream
 
 /**
  * Emits one record per ECG waveform sample. Sample i is timed at the reading start plus
@@ -29,7 +32,7 @@ import java.time.Instant
  * metadata (heart rate, sampling parameters, device info) is repeated on every sample's record,
  * linked by the shared reading id.
  */
-class ElectrocardiogramGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
+class GoogleHealthElectrocardiogramAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,
         user: User,
@@ -38,16 +41,18 @@ class ElectrocardiogramGoogleHealthAvroConverter(topic: String) : GoogleHealthAv
         val start = data["interval"]?.get("startTime")?.asText()
             ?.let { runCatching { Instant.parse(it) }.getOrNull() } ?: return emptyList()
         val id = (point["name"] ?: point["dataPointName"])?.asText()?.substringAfterLast('/')
-            ?: run {
-                logger.warn("Dropping electrocardiogram data point with no usable id for user={}", user.versionedId)
-                return emptyList()
-            }
+            ?: throw IOException(
+                "Electrocardiogram data point has no name or dataPointName to derive an id from " +
+                    "for user=${user.versionedId}",
+            )
         val samples = data["waveformSamples"]?.takeIf { it.isArray } ?: return emptyList()
-        val frequency = data["samplingFrequencyHertz"]?.takeIf { !it.isNull }?.asInt()?.takeIf { it > 0 }
+        val frequency = data["samplingFrequencyHertz"]?.takeIf { !it.isNull }?.asInt()
+            ?.takeIf { it > 0 }
             ?: return emptyList()
 
         val device = data["medicalDeviceInfo"]
-        val beatsPerMinuteAvg = data["beatsPerMinuteAvg"]?.takeIf { !it.isNull }?.asText()?.toIntOrNull()
+        val beatsPerMinuteAvg = data["beatsPerMinuteAvg"]?.takeIf { !it.isNull }?.asText()
+            ?.toIntOrNull()
         val scalingFactor = data["millivoltsScalingFactor"]?.takeIf { !it.isNull }?.asInt()
         val leadNumber = data["leadNumber"]?.takeIf { !it.isNull }?.asInt()
         val deviceModel = device?.get("deviceModel")?.asText()
@@ -56,12 +61,19 @@ class ElectrocardiogramGoogleHealthAvroConverter(topic: String) : GoogleHealthAv
 
         val startSec = epochSeconds(start)
         val received = nowEpochSeconds()
-        return samples.mapIndexed { i, sampleNode ->
+        val sampleCount = samples.size()
+        if (sampleCount == 0) return emptyList()
+
+        // A single reading carries thousands of samples (~7500 at 30 s / 250 Hz), each needing
+        // its own record. Every iteration only reads the parsed JSON tree and builds an
+        // independent Avro record, so the work spreads safely over the common pool; the stream
+        // stays ordered, keeping the samples in acquisition order.
+        return IntStream.range(0, sampleCount).parallel().mapToObj { i ->
             val record = googleHealthElectrocardiogram {
                 time = startSec + i.toDouble() / frequency
                 timeReceived = received
                 this.id = id
-                sample = sampleNode.asInt()
+                sample = samples[i].asInt()
                 this.beatsPerMinuteAvg = beatsPerMinuteAvg
                 this.samplingFrequencyHertz = frequency
                 this.millivoltsScalingFactor = scalingFactor
@@ -71,6 +83,6 @@ class ElectrocardiogramGoogleHealthAvroConverter(topic: String) : GoogleHealthAv
                 this.featureVersion = featureVersion
             }
             user.observationKey to record
-        }
+        }.collect(Collectors.toList())
     }
 }

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/ExerciseGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthExerciseAvroConverter.kt

@@ -19,12 +19,13 @@ package org.radarbase.googlehealth.converter
 import com.fasterxml.jackson.databind.JsonNode
 import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
-import org.radarbase.googlehealth.util.googleHealthExerciseHeartRate
 import org.radarbase.googlehealth.util.googleHealthExercise
+import org.radarbase.googlehealth.util.googleHealthExerciseHeartRate
 import org.radarbase.googlehealth.util.googleHealthSource
+import java.io.IOException
 import java.time.Instant
 
-class ExerciseGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
+class GoogleHealthExerciseAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,
         user: User,
@@ -44,21 +45,27 @@ class ExerciseGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConvert
         val distanceKm = metrics?.get("distanceMillimeters")?.takeIf { !it.isNull }
             ?.asText()?.toDoubleOrNull()?.let { it.toFloat() / 1_000_000f }
 
-        val caloriesKcal = metrics?.get("caloriesKcal")?.takeIf { !it.isNull }?.asText()?.toDoubleOrNull()
+        val caloriesKcal = metrics?.get(
+            "caloriesKcal",
+        )?.takeIf { !it.isNull }?.asText()?.toDoubleOrNull()
         val energyKj = caloriesKcal?.let { (it * KCAL_TO_KJ).toFloat() }
         val stepCount = metrics?.get("steps")?.takeIf { !it.isNull }?.asText()?.toIntOrNull()
 
         val speedKmh = metrics?.get("averageSpeedMillimetersPerSecond")?.takeIf { !it.isNull }
             ?.asText()?.toDoubleOrNull()?.let { it * MM_PER_S_TO_KM_PER_H }
 
-        val avgHr = metrics?.get("averageHeartRateBeatsPerMinute")?.takeIf { !it.isNull }?.asText()?.toIntOrNull()
+        val avgHr = metrics?.get("averageHeartRateBeatsPerMinute")?.takeIf {
+            !it.isNull
+        }?.asText()?.toIntOrNull()
         val zones = metrics?.get("heartRateZoneDurations")?.takeIf { !it.isNull }
         val avgHeartRate = if (avgHr != null || zones != null) {
             googleHealthExerciseHeartRate {
                 mean = avgHr
                 durationLight = zones?.get("lightTime")?.asText()?.let { parseDurationSeconds(it) }
-                durationModerate = zones?.get("moderateTime")?.asText()?.let { parseDurationSeconds(it) }
-                durationVigorous = zones?.get("vigorousTime")?.asText()?.let { parseDurationSeconds(it) }
+                durationModerate = zones?.get("moderateTime")?.asText()
+                    ?.let { parseDurationSeconds(it) }
+                durationVigorous = zones?.get("vigorousTime")?.asText()
+                    ?.let { parseDurationSeconds(it) }
                 durationPeak = zones?.get("peakTime")?.asText()?.let { parseDurationSeconds(it) }
             }
         } else {
@@ -76,12 +83,17 @@ class ExerciseGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConvert
             }
         }
 
-        val activityId = (point["name"] ?: point["dataPointName"])?.asText()
-            ?.substringAfterLast('/')?.toLongOrNull()
-            ?: run {
-                logger.warn("Dropping exercise data point with no usable log id for user={}", user.versionedId)
-                return emptyList()
-            }
+        val idSegment = (point["name"] ?: point["dataPointName"])?.asText()
+            ?.substringAfterLast('/')
+            ?: throw IOException(
+                "Exercise data point has no name or dataPointName to derive a log id from " +
+                    "for user=${user.versionedId}",
+            )
+        val activityId = idSegment.toLongOrNull()
+            ?: throw IOException(
+                "Exercise data point log id '$idSegment' is not numeric " +
+                    "for user=${user.versionedId}",
+            )
 
         val record = googleHealthExercise {
             time = epochSeconds(start)

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthFloorsAvroConverter.kt

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2026 King's College London
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.radarbase.googlehealth.converter
+
+import com.fasterxml.jackson.databind.JsonNode
+import org.apache.avro.specific.SpecificRecord
+import org.radarbase.googlehealth.user.User
+import org.radarbase.googlehealth.util.googleHealthFloors
+
+/**
+ * Converts `floors` data points: elevation gained over an interval.
+ * Google documents `count` as an int64, serialised as a JSON string, `asInt` parses either form.
+ * The type supports true zeros, so a `count` of 0 is a real observation and is kept.
+ */
+class GoogleHealthFloorsAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
+    override fun convertDataPoint(
+        point: JsonNode,
+        user: User,
+    ): List<Pair<SpecificRecord, SpecificRecord>> {
+        val data = point["floors"] ?: return emptyList()
+        val (start, end) = parseInterval(data) ?: return emptyList()
+        val count = data["count"]?.asInt() ?: return emptyList()
+        val record = googleHealthFloors {
+            time = epochSeconds(start)
+            endTime = epochSeconds(end)
+            timeReceived = nowEpochSeconds()
+            floors = count
+        }
+        return listOf(user.observationKey to record)
+    }
+}

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/HeartRateGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthHeartRateAvroConverter.kt

@@ -21,7 +21,7 @@ import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthHeartRate
 
-class HeartRateGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
+class GoogleHealthHeartRateAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,
         user: User,

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/HeartRateVariabilityGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthHeartRateVariabilityAvroConverter.kt

@@ -21,15 +21,17 @@ import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthHeartRateVariability
 
-class HeartRateVariabilityGoogleHealthAvroConverter(topic: String) :
+class GoogleHealthHeartRateVariabilityAvroConverter(topic: String) :
     GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,
         user: User,
     ): List<Pair<SpecificRecord, SpecificRecord>> {
         val data = point["heartRateVariability"] ?: return emptyList()
         val time = parseSampleTime(data) ?: return emptyList()
-        val rmssd = data["rootMeanSquareOfSuccessiveDifferencesMilliseconds"]?.takeIf { it.isNumber }?.floatValue()
+        val rmssd = data["rootMeanSquareOfSuccessiveDifferencesMilliseconds"]?.takeIf {
+            it.isNumber
+        }?.floatValue()
             ?: return emptyList()
         val record = googleHealthHeartRateVariability {
             this.time = epochSeconds(time)

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/IrregularRhythmNotificationGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthIrregularRhythmNotificationAvroConverter.kt

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.JsonNode
 import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthIrregularRhythmNotification
+import java.io.IOException
 import java.time.Instant
 
 /**
@@ -28,17 +29,19 @@ import java.time.Instant
  * plus the context of its parent window (start/end, positive) and session (start), with the
  * device metadata repeated, linked by the shared notification id.
  */
-class IrregularRhythmNotificationGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
+class GoogleHealthIrregularRhythmNotificationAvroConverter(
+    topic: String,
+) : GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,
         user: User,
     ): List<Pair<SpecificRecord, SpecificRecord>> {
         val data = point["irregularRhythmNotification"] ?: return emptyList()
         val id = (point["name"] ?: point["dataPointName"])?.asText()?.substringAfterLast('/')
-            ?: run {
-                logger.warn("Dropping irregularRhythmNotification data point with no usable id for user={}", user.versionedId)
-                return emptyList()
-            }
+            ?: throw IOException(
+                "Irregular rhythm notification data point has no name or dataPointName to derive " +
+                    "an id from for user=${user.versionedId}",
+            )
         val windows = data["alertWindows"]?.takeIf { it.isArray } ?: return emptyList()
 
         val device = data["medicalDeviceInfo"]
@@ -56,11 +59,13 @@ class IrregularRhythmNotificationGoogleHealthAvroConverter(topic: String) : Goog
                 ?.let { runCatching { Instant.parse(it) }.getOrNull() }
             if (windowStart == null || windowEnd == null) return@flatMap emptyList()
             val positive = window["positive"]?.takeIf { !it.isNull }?.asBoolean()
-            val heartBeats = window["heartBeats"]?.takeIf { it.isArray } ?: return@flatMap emptyList()
+            val heartBeats = window["heartBeats"]?.takeIf { it.isArray }
+                ?: return@flatMap emptyList()
 
             heartBeats.mapNotNull { beat ->
                 val beatTime = (beat["physicalTime"] ?: beat["time"])?.asText()
-                    ?.let { runCatching { Instant.parse(it) }.getOrNull() } ?: return@mapNotNull null
+                    ?.let { runCatching { Instant.parse(it) }.getOrNull() }
+                    ?: return@mapNotNull null
                 val record = googleHealthIrregularRhythmNotification {
                     time = epochSeconds(beatTime)
                     timeReceived = received
@@ -69,7 +74,8 @@ class IrregularRhythmNotificationGoogleHealthAvroConverter(topic: String) : Goog
                     windowStartTime = epochSeconds(windowStart)
                     windowEndTime = epochSeconds(windowEnd)
                     this.positive = positive
-                    beatsPerMinute = beat["beatsPerMinute"]?.takeIf { !it.isNull }?.asText()?.toIntOrNull()
+                    beatsPerMinute = beat["beatsPerMinute"]?.takeIf { !it.isNull }?.asText()
+                        ?.toIntOrNull()
                     this.firmwareVersion = firmwareVersion
                     this.featureVersion = featureVersion
                     this.deviceModel = deviceModel

google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/OxygenSaturationGoogleHealthAvroConverter.kt → google-health-library/src/main/kotlin/org/radarbase/googlehealth/converter/GoogleHealthOxygenSaturationAvroConverter.kt

@@ -21,7 +21,7 @@ import org.apache.avro.specific.SpecificRecord
 import org.radarbase.googlehealth.user.User
 import org.radarbase.googlehealth.util.googleHealthOxygenSaturation
 
-class OxygenSaturationGoogleHealthAvroConverter(topic: String) :
+class GoogleHealthOxygenSaturationAvroConverter(topic: String) :
     GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
         point: JsonNode,