Add converters for ECG and IRN data, and update exercise data parsing.

google-health-library/build.gradle

@@ -1,6 +1,6 @@
 
 group = 'org.radarbase'
-version = '0.0.1'
+version = '1.0.0'
 
 apply plugin: 'maven-publish'
 

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

@@ -0,0 +1,76 @@
+/*
+ * 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.googleHealthElectrocardiogram
+import java.time.Instant
+
+/**
+ * Emits one record per ECG waveform sample. Sample i is timed at the reading start plus
+ * i / samplingFrequencyHertz seconds and carries the raw waveform value as reported by the
+ * device (an ADC count; divide by millivoltsScalingFactor for millivolts). The reading-level
+ * 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) {
+    override fun convertDataPoint(
+        point: JsonNode,
+        user: User,
+    ): List<Pair<SpecificRecord, SpecificRecord>> {
+        val data = point["electrocardiogram"] ?: return emptyList()
+        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()
+            }
+        val samples = data["waveformSamples"]?.takeIf { it.isArray } ?: return emptyList()
+        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 scalingFactor = data["millivoltsScalingFactor"]?.takeIf { !it.isNull }?.asInt()
+        val leadNumber = data["leadNumber"]?.takeIf { !it.isNull }?.asInt()
+        val deviceModel = device?.get("deviceModel")?.asText()
+        val firmwareVersion = device?.get("firmwareVersion")?.asText()
+        val featureVersion = device?.get("featureVersion")?.asText()
+
+        val startSec = epochSeconds(start)
+        val received = nowEpochSeconds()
+        return samples.mapIndexed { i, sampleNode ->
+            val record = googleHealthElectrocardiogram {
+                time = startSec + i.toDouble() / frequency
+                timeReceived = received
+                this.id = id
+                sample = sampleNode.asInt()
+                this.beatsPerMinuteAvg = beatsPerMinuteAvg
+                this.samplingFrequencyHertz = frequency
+                this.millivoltsScalingFactor = scalingFactor
+                this.leadNumber = leadNumber
+                this.deviceModel = deviceModel
+                this.firmwareVersion = firmwareVersion
+                this.featureVersion = featureVersion
+            }
+            user.observationKey to record
+        }
+    }
+}

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

@@ -19,8 +19,10 @@ 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.exerciseHeartRate
-import org.radarbase.googlehealth.util.activityLogRecord
+import org.radarbase.googlehealth.util.googleHealthExerciseHeartRate
+import org.radarbase.googlehealth.util.googleHealthExercise
+import org.radarbase.googlehealth.util.googleHealthSource
+import java.time.Instant
 
 class ExerciseGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConverter(topic) {
     override fun convertDataPoint(
@@ -33,45 +35,77 @@ class ExerciseGoogleHealthAvroConverter(topic: String) : GoogleHealthAvroConvert
             data["interval"]?.get("startUtcOffset")?.asText(),
         )
         val durationSec = (end.epochSecond - start.epochSecond).toFloat().coerceAtLeast(0.0f)
+        val activeDurationSec = data["activeDuration"]?.asText()
+            ?.let { parseDurationSeconds(it).toFloat() } ?: durationSec
+        val lastModified = data["updateTime"]?.asText()
+            ?.let { runCatching { Instant.parse(it) }.getOrNull() } ?: end
         val metrics = data["metricsSummary"]
+
         val distanceKm = metrics?.get("distanceMillimeters")?.takeIf { !it.isNull }
-            ?.asDouble()?.let { it.toFloat() / 1_000_000f }
-        val caloriesKcal = metrics?.get("caloriesKcal")?.takeIf { !it.isNull }?.asDouble()
+            ?.asText()?.toDoubleOrNull()?.let { it.toFloat() / 1_000_000f }
+
+        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 }?.asInt()
-        val avgHr = metrics?.get("averageHeartRateBeatsPerMinute")?.takeIf { !it.isNull }?.asInt()
-        val avgHeartRate = avgHr?.let { exerciseHeartRate { mean = it } }
+        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 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) }
+                durationPeak = zones?.get("peakTime")?.asText()?.let { parseDurationSeconds(it) }
+            }
+        } else {
+            null
+        }
         val exerciseType = data["exerciseType"]?.asText()
-        // The exercise (log) id is the last segment of the reconcile data point's `dataPointName`
-        // (e.g. users/{u}/dataTypes/exercise/dataPoints/7726011858216679720). Exercise is an
-        // identifiable data type, so this is always present — fail loudly rather than emit a
-        // fabricated id. It is also the id used to export the session's TCX track.
-        val activityId = point["dataPointName"]?.asText()?.substringAfterLast('/')?.toLongOrNull()
-            ?: throw IllegalStateException("Exercise data point has no usable dataPointName log id: $point")
-        val record = activityLogRecord {
+        val dataSource = point["dataSource"]?.takeIf { !it.isNull }
+        val device = dataSource?.get("device")
+        val exerciseSource = dataSource?.let {
+            googleHealthSource {
+                name = device?.get("displayName")?.asText()
+                formFactor = device?.get("formFactor")?.asText()
+                manufacturer = device?.get("manufacturer")?.asText()
+                platform = it["platform"]?.asText()
+            }
+        }
+
+        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 record = googleHealthExercise {
             time = epochSeconds(start)
             timeReceived = nowEpochSeconds()
             timeZoneOffset = offsetSeconds
-            timeLastModified = epochSeconds(end)
+            timeLastModified = epochSeconds(lastModified)
             duration = durationSec
-            durationActive = durationSec
+            durationActive = activeDurationSec
             id = activityId
             name = data["displayName"]?.asText() ?: exerciseType
-            logType = point["dataSource"]?.get("recordingMethod")?.asText()
-            type = null
-            source = null
-            manualDataEntry = null
+            logType = dataSource?.get("recordingMethod")?.asText()
+            type = exerciseType
+            source = exerciseSource
             energy = energyKj
-            levels = null
             heartRate = avgHeartRate
             steps = stepCount
             distance = distanceKm
-            speed = null
+            speed = speedKmh
         }
         return listOf(user.observationKey to record)
     }
 
     companion object {
         private const val KCAL_TO_KJ = 4.1868
+        private const val MM_PER_S_TO_KM_PER_H = 0.0036
     }
 }

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

@@ -19,6 +19,8 @@ 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.slf4j.Logger
+import org.slf4j.LoggerFactory
 import java.time.Instant
 import java.time.LocalDate
 import java.time.LocalDateTime
@@ -27,6 +29,8 @@ import java.time.ZoneOffset
 
 abstract class GoogleHealthAvroConverter(override val topic: String) : AvroConverter {
 
+    protected val logger: Logger = LoggerFactory.getLogger(javaClass)
+
     abstract fun convertDataPoint(
         point: JsonNode,
         user: User,
@@ -79,13 +83,17 @@ abstract class GoogleHealthAvroConverter(override val topic: String) : AvroConve
             return dateTime.toInstant(ZoneOffset.ofTotalSeconds(utcOffsetSeconds))
         }
 
-        fun parseUtcOffsetSeconds(durationText: String?): Int {
+        /** Parses a Google `Duration` string (e.g. "36s", "780s") to whole seconds. */
+        fun parseDurationSeconds(durationText: String?): Int {
             if (durationText.isNullOrEmpty()) return 0
             val trimmed = durationText.trim().removeSuffix("s")
-            val seconds = trimmed.toLongOrNull() ?: return 0
+            val seconds = trimmed.toDoubleOrNull() ?: return 0
             return seconds.toInt()
         }
 
+        /** A UTC offset is encoded as a [Duration]; returns its value in seconds. */
+        fun parseUtcOffsetSeconds(offsetText: String?): Int = parseDurationSeconds(offsetText)
+
         fun epochSeconds(instant: Instant): Double = instant.toEpochMilli() / 1000.0
 
         fun nowEpochSeconds(): Double = Instant.now().toEpochMilli() / 1000.0

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

@@ -29,7 +29,7 @@ class HeartRateVariabilityGoogleHealthAvroConverter(topic: String) :
     ): List<Pair<SpecificRecord, SpecificRecord>> {
         val data = point["heartRateVariability"] ?: return emptyList()
         val time = parseSampleTime(data) ?: return emptyList()
-        val rmssd = data["rootMeanSquareOfSuccessiveDifferencesMilliseconds"]?.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

@@ -0,0 +1,81 @@
+/*
+ * 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.googleHealthIrregularRhythmNotification
+import java.time.Instant
+
+/**
+ * Emits one record per heart beat of an Irregular Rhythm Notification, flattening the API's
+ * session -> alertWindow -> heartBeat hierarchy. Each record carries the heart beat's own time
+ * 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) {
+    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()
+            }
+        val windows = data["alertWindows"]?.takeIf { it.isArray } ?: return emptyList()
+
+        val device = data["medicalDeviceInfo"]
+        val sessionStart = data["interval"]?.get("startTime")?.asText()
+            ?.let { runCatching { Instant.parse(it) }.getOrNull() }
+        val firmwareVersion = device?.get("firmwareVersion")?.asText()
+        val featureVersion = device?.get("featureVersion")?.asText()
+        val deviceModel = device?.get("deviceModel")?.asText()
+        val received = nowEpochSeconds()
+
+        return windows.flatMap { window ->
+            val windowStart = window["startTime"]?.asText()
+                ?.let { runCatching { Instant.parse(it) }.getOrNull() }
+            val windowEnd = window["endTime"]?.asText()
+                ?.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()
+
+            heartBeats.mapNotNull { beat ->
+                val beatTime = (beat["physicalTime"] ?: beat["time"])?.asText()
+                    ?.let { runCatching { Instant.parse(it) }.getOrNull() } ?: return@mapNotNull null
+                val record = googleHealthIrregularRhythmNotification {
+                    time = epochSeconds(beatTime)
+                    timeReceived = received
+                    this.id = id
+                    sessionStartTime = sessionStart?.let { epochSeconds(it) }
+                    windowStartTime = epochSeconds(windowStart)
+                    windowEndTime = epochSeconds(windowEnd)
+                    this.positive = positive
+                    beatsPerMinute = beat["beatsPerMinute"]?.takeIf { !it.isNull }?.asText()?.toIntOrNull()
+                    this.firmwareVersion = firmwareVersion
+                    this.featureVersion = featureVersion
+                    this.deviceModel = deviceModel
+                }
+                user.observationKey to record
+            }
+        }
+    }
+}

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

@@ -29,7 +29,7 @@ class OxygenSaturationGoogleHealthAvroConverter(topic: String) :
     ): List<Pair<SpecificRecord, SpecificRecord>> {
         val data = point["oxygenSaturation"] ?: return emptyList()
         val time = parseSampleTime(data) ?: return emptyList()
-        val pct = data["percentage"]?.floatValue() ?: return emptyList()
+        val pct = data["percentage"]?.takeIf { it.isNumber }?.floatValue() ?: return emptyList()
         val record = googleHealthOxygenSaturation {
             this.time = epochSeconds(time)
             timeReceived = nowEpochSeconds()

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

@@ -29,22 +29,21 @@ class RespiratoryRateSleepSummaryGoogleHealthAvroConverter(topic: String) :
     ): List<Pair<SpecificRecord, SpecificRecord>> {
         val data = point["respiratoryRateSleepSummary"] ?: return emptyList()
         val time = parseSampleTime(data) ?: return emptyList()
-        val deep = data["deepSleepStats"]?.get("breathsPerMinute")?.floatValue() ?: UNAVAILABLE
-        val full = data["fullSleepStats"]?.get("breathsPerMinute")?.floatValue() ?: UNAVAILABLE
-        val light = data["lightSleepStats"]?.get("breathsPerMinute")?.floatValue() ?: UNAVAILABLE
-        val rem = data["remSleepStats"]?.get("breathsPerMinute")?.floatValue() ?: UNAVAILABLE
+        val deep = data["deepSleepStats"]?.get("breathsPerMinute")
+        val full = data["fullSleepStats"]?.get("breathsPerMinute")
+        val light = data["lightSleepStats"]?.get("breathsPerMinute")
+        val rem = data["remSleepStats"]?.get("breathsPerMinute")
+
+        if (listOf(deep, full, light, rem).any { it != null && !it.isNumber }) return emptyList()
+
         val record = googleHealthRespiratoryRateSleepSummary {
             this.time = epochSeconds(time)
             timeReceived = nowEpochSeconds()
-            lightSleep = light
-            deepSleep = deep
-            remSleep = rem
-            fullSleep = full
+            lightSleep = light?.floatValue()
+            deepSleep = deep?.floatValue()
+            remSleep = rem?.floatValue()
+            fullSleep = full?.floatValue()
         }
         return listOf(user.observationKey to record)
     }
-
-    companion object {
-        private const val UNAVAILABLE = 0.0f
-    }
 }