Skip to content

Stream Health Connect reads page-by-page + optional native downsampling (fixes HR OOM) #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dmarttila merged 2 commits into from
Jul 14, 2026
Merged

Stream Health Connect reads page-by-page + optional native downsampling (fixes HR OOM) #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fd85060
Convert Health Connect records page-by-page and add optional native d…
oliviawalch Jul 12, 2026
7882f9e
Aggregate heart rate as BPM_AVG per bucket; harden getAggregateHCData
oliviawalch Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
174 changes: 116 additions & 58 deletions packages/health/android/src/main/kotlin/cachet/plugins/health/HealthPlugin.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2966,19 +2966,75 @@ class HealthPlugin(private var channel: MethodChannel? = null) :
val dataType = call.argument<String>("dataTypeKey")!!
val startTime = Instant.ofEpochMilli(call.argument<Long>("startTime")!!)
val endTime = Instant.ofEpochMilli(call.argument<Long>("endTime")!!)
// Optional minimum spacing between returned samples, in milliseconds.
// When set, at most one converted data point is kept per interval.
// (The value fits in an Int for small intervals, so read it as Number.)
val samplingIntervalMillis =
call.argument<Number>("samplingIntervalMillis")?.toLong()
val healthConnectData = mutableListOf<Map<String, Any?>>()

scope.launch {
try {
MapToHCType[dataType]?.let { classType ->
val records = mutableListOf<Record>()

val granted = healthConnectClient.permissionController.getGrantedPermissions()
if (!granted.contains(HealthPermission.getReadPermission(classType))) {
Log.d("Health Plugin", "No Permission granted for $dataType")
return@let
}

// Workouts and sleep sessions have special conversion logic
// below that needs the full record objects; they are
// low-volume, so reading them all into memory is safe.
// Every other type is converted one page at a time so that
// dense histories (heart rate in particular can have a
// sample every few seconds) are never fully materialized
// as raw records + converted maps at once — doing so
// caused OutOfMemoryError crashes for heavy users.
if (dataType != WORKOUT && classType != SleepSessionRecord::class) {
// Timestamp (epoch millis) of the last data point kept
// by the sampling filter, across page boundaries.
var lastKeptMillis: Long? = null
var pageToken: String? = null
do {
val response =
healthConnectClient.readRecords(
ReadRecordsRequest(
recordType = classType,
timeRangeFilter =
TimeRangeFilter.between(
startTime,
endTime
),
pageToken = pageToken,
)
)
for (rec in response.records) {
for (converted in convertRecord(rec, dataType)) {
if (samplingIntervalMillis != null) {
val dateFrom =
(converted["date_from"] as? Number)?.toLong()
if (dateFrom != null) {
val last = lastKeptMillis
if (last != null &&
dateFrom - last < samplingIntervalMillis
) {
// Too close to the previously
// kept sample — drop it.
continue
}
lastKeptMillis = dateFrom
}
}
healthConnectData.add(converted)
}
}
pageToken = response.pageToken
} while (!pageToken.isNullOrEmpty())
return@let
}

val records = mutableListOf<Record>()

// Set up the initial request to read health records with specified
// parameters
var request =
Expand Down Expand Up @@ -3165,12 +3221,6 @@ class HealthPlugin(private var channel: MethodChannel? = null) :
}
}
}
} else {
for (rec in records) {
healthConnectData.addAll(
convertRecord(rec, dataType)
)
}
}
}

Expand Down Expand Up @@ -3209,59 +3259,63 @@ class HealthPlugin(private var channel: MethodChannel? = null) :
val endTime = Instant.ofEpochMilli(call.argument<Long>("endTime")!!)
val healthConnectData = mutableListOf<Map<String, Any?>>()
scope.launch {
MapToHCAggregateMetric[dataType]?.let { metricClassType ->
val request =
AggregateGroupByDurationRequest(
metrics = setOf(metricClassType),
timeRangeFilter =
TimeRangeFilter.between(
startTime,
endTime
),
timeRangeSlicer =
Duration.ofSeconds(
interval
try {
MapToHCAggregateMetric[dataType]?.let { metricClassType ->
val request =
AggregateGroupByDurationRequest(
metrics = setOf(metricClassType),
timeRangeFilter =
TimeRangeFilter.between(
startTime,
endTime
),
timeRangeSlicer =
Duration.ofSeconds(
interval
)
)
)
val response = healthConnectClient.aggregateGroupByDuration(request)

for (durationResult in response) {
// The result may be null if no data is available in the
// time range
var totalValue = durationResult.result[metricClassType]
if (totalValue is Length) {
totalValue = totalValue.inMeters
} else if (totalValue is Energy) {
totalValue = totalValue.inKilocalories
}
val response = healthConnectClient.aggregateGroupByDuration(request)

for (durationResult in response) {
// The result is null if no data is available in the
// bucket — skip it rather than fabricating a zero
// value (a 0 BPM heart rate is not a real data point).
var totalValue = durationResult.result[metricClassType]
?: continue
if (totalValue is Length) {
totalValue = totalValue.inMeters
} else if (totalValue is Energy) {
totalValue = totalValue.inKilocalories
}

val packageNames =
durationResult.result.dataOrigins
.joinToString { origin ->
"${origin.packageName}"
}
val packageNames =
durationResult.result.dataOrigins
.joinToString { origin ->
"${origin.packageName}"
}

val data =
mapOf<String, Any>(
"value" to
(totalValue
?: 0),
"date_from" to
durationResult.startTime
.toEpochMilli(),
"date_to" to
durationResult.endTime
.toEpochMilli(),
"source_name" to
packageNames,
"source_id" to "",
"is_manual_entry" to
packageNames.contains(
"user_input"
)
)
healthConnectData.add(data)
val data =
mapOf<String, Any>(
"value" to totalValue,
"date_from" to
durationResult.startTime
.toEpochMilli(),
"date_to" to
durationResult.endTime
.toEpochMilli(),
"source_name" to
packageNames,
"source_id" to "",
"is_manual_entry" to
packageNames.contains(
"user_input"
)
)
healthConnectData.add(data)
}
}
} catch (e: Exception) {
Log.e("FLUTTER_HEALTH", e.toString())
}
Handler(context!!.mainLooper).run { result.success(healthConnectData) }
}
Expand Down Expand Up @@ -4468,7 +4522,11 @@ class HealthPlugin(private var channel: MethodChannel? = null) :
ACTIVE_ENERGY_BURNED to
ActiveCaloriesBurnedRecord
.ACTIVE_CALORIES_TOTAL,
HEART_RATE to HeartRateRecord.MEASUREMENTS_COUNT,
// BPM_AVG, not MEASUREMENTS_COUNT: interval queries for heart
// rate should return the average BPM per bucket, which lets
// callers read dense HR histories without ever materializing the
// raw samples (the aggregation runs inside Health Connect).
HEART_RATE to HeartRateRecord.BPM_AVG,
DISTANCE_DELTA to DistanceRecord.DISTANCE_TOTAL,
WATER to HydrationRecord.VOLUME_TOTAL,
SLEEP_ASLEEP to SleepSessionRecord.SLEEP_DURATION_TOTAL,
Expand Down
28 changes: 20 additions & 8 deletions packages/health/lib/src/health_plugin.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -622,17 +622,23 @@ class Health {
}

/// Fetch a list of health data points based on [types].
///
/// If [samplingInterval] is provided, at most one data point is returned
/// per interval, dropping the rest as the data is read. Use this for very
/// dense types (e.g. heart rate) where reading every raw sample can exhaust
/// memory. Currently only applied on Android (Health Connect).
Future<List<HealthDataPoint>> getHealthDataFromTypes({
required List<HealthDataType> types,
required DateTime startTime,
required DateTime endTime,
bool includeManualEntry = true,
Duration? samplingInterval,
}) async {
List<HealthDataPoint> dataPoints = [];

for (var type in types) {
final result =
await _prepareQuery(startTime, endTime, type, includeManualEntry);
final result = await _prepareQuery(
startTime, endTime, type, includeManualEntry, samplingInterval);
dataPoints.addAll(result);
}

Expand Down Expand Up @@ -684,8 +690,9 @@ class Health {
DateTime startTime,
DateTime endTime,
HealthDataType dataType,
bool includeManualEntry,
) async {
bool includeManualEntry, [
Duration? samplingInterval,
]) async {
// Ask for device ID only once
_deviceId ??= Platform.isAndroid
? (await _deviceInfo.androidInfo).id
Expand All @@ -701,7 +708,8 @@ class Health {
if (dataType == HealthDataType.BODY_MASS_INDEX && Platform.isAndroid) {
return _computeAndroidBMI(startTime, endTime, includeManualEntry);
}
return await _dataQuery(startTime, endTime, dataType, includeManualEntry);
return await _dataQuery(
startTime, endTime, dataType, includeManualEntry, samplingInterval);
}

/// Prepares an interval query, i.e. checks if the types are available, etc.
Expand Down Expand Up @@ -750,14 +758,18 @@ class Health {
}

/// Fetches data points from Android/iOS native code.
Future<List<HealthDataPoint>> _dataQuery(DateTime startTime, DateTime endTime,
HealthDataType dataType, bool includeManualEntry) async {
Future<List<HealthDataPoint>> _dataQuery(
DateTime startTime, DateTime endTime, HealthDataType dataType,
bool includeManualEntry,
[Duration? samplingInterval]) async {
final args = <String, dynamic>{
'dataTypeKey': dataType.name,
'dataUnitKey': dataTypeToUnit[dataType]!.name,
'startTime': startTime.millisecondsSinceEpoch,
'endTime': endTime.millisecondsSinceEpoch,
'includeManualEntry': includeManualEntry
'includeManualEntry': includeManualEntry,
if (samplingInterval != null)
'samplingIntervalMillis': samplingInterval.inMilliseconds,
};
final fetchedDataPoints = await _channel.invokeMethod('getData', args);

Expand Down
Loading