Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
23 changes: 11 additions & 12 deletions lib/ble/ble_engine.dart
Original file line number Diff line number Diff line change
Expand Up @@ -2965,7 +2965,7 @@ class BleEngine {
// just stores directly if a frame somehow arrives before setup completed.
final d = _drain;
if (d != null) {
d.onHistoricalRecord(raw, sample);
d.onHistoricalRecord(raw, sample, recType);
} else {
unawaited(_storeRecord(sample, raw));
}
Expand Down Expand Up @@ -5015,11 +5015,13 @@ class DrainController {
int get currentBurstHistoricalPacketCount => burstStats.historicalPacketCount;
String get currentBurstBreakdown => burstStats.breakdownString;

void onHistoricalRecord(RawRecord raw, Sample? sample) {
/// [revision] is the record version byte the ingest path already read off
/// the frame (-1 when the frame was too short to have one).
void onHistoricalRecord(RawRecord raw, Sample? sample, int revision) {
records++;
recordsThisOffload++;
_lastProgressAt = DateTime.now();
burstStats.onHistoricalData(raw.packetType, raw.counter, sample, raw.hex);
burstStats.onHistoricalData(raw.packetType, raw.counter, revision);
if (_buffering) {
_raws.add(raw);
_samples.add(sample);
Expand Down Expand Up @@ -5364,19 +5366,16 @@ class BurstStats {
return parts.join(', ');
}

void onHistoricalData(
int packetType,
int counter,
Sample? sample,
String rawHex,
) {
/// [revision] is the record version byte (inner[1]), which the caller has
/// already read off the frame. This used to take the record's hex and parse
/// the whole thing back into bytes to reach that one byte — a throwaway
/// buffer per record, on every record of every offload.
void onHistoricalData(int packetType, int counter, int revision) {
if (packetType != PacketType.historicalData) return;
final inner = hexToBytes(rawHex);
if (inner.length < 2) {
if (revision < 0) {
_unknownCount++;
return;
}
final revision = inner[1];
if (_ordinaryHistoricalRevisions.contains(revision)) {
_dataPacketCountsByRevision[revision] =
(_dataPacketCountsByRevision[revision] ?? 0) + 1;
Expand Down
83 changes: 0 additions & 83 deletions lib/compute/derivation_engine.dart
Original file line number Diff line number Diff line change
Expand Up @@ -3172,8 +3172,6 @@ class DerivationEngine {
sleepHr: sleepSub.hr,
sleepRrTsMs: sleepSub.rrTsMs,
sleepRrMs: sleepSub.rrMs,
sleepSpo2Red: sleepSub.spo2Red,
sleepSpo2Ir: sleepSub.spo2Ir,
sleepSkinTemp: sleepSub.skinTemp,
sleepJson: day.sleepJson,
hypnoStages: day.hypnoStages,
Expand All @@ -3200,7 +3198,6 @@ class DerivationEngine {
_perDayTimeout,
label: 'day-bundle ${day.date}',
);
_logSpo2Diagnostics(day, input, bundle);
// Readiness came back absent for TODAY specifically (not a historical
// backfill day, which would just be noise) — log why. This ran inside
// Isolate.run so it couldn't call Firebase itself; it just returned the
Expand Down Expand Up @@ -3722,86 +3719,6 @@ class DerivationEngine {
_log('froze headline readiness ${next.value} for ${next.day}');
}

void _logSpo2Diagnostics(
PreparedDerivationDay day,
DayBundleInput input,
Map<String, dynamic> bundle,
) {
final red = input.sleepSpo2Red;
final ir = input.sleepSpo2Ir;
final ts = input.sleepTsSec;
if (red.isEmpty || ir.isEmpty || ts.isEmpty) {
_log('[spo2-detect] {"day":"${day.date}","status":"no_sleep_spo2"}');
return;
}

int minInt(List<int> xs) => xs.reduce((a, b) => a < b ? a : b);
int maxInt(List<int> xs) => xs.reduce((a, b) => a > b ? a : b);
double meanInt(List<int> xs) =>
xs.isEmpty ? 0 : xs.reduce((a, b) => a + b) / xs.length;

final redNonZero = red.where((v) => v > 0).length;
final irNonZero = ir.where((v) => v > 0).length;
final spo2 = (bundle['spo2'] as Map?)?.cast<String, dynamic>();
final ratios = <double>[
for (var i = 0; i < red.length && i < ir.length; i++)
if (red[i] > 0 && ir[i] > 0) red[i] / ir[i],
];
double? meanDouble(List<double> xs) =>
xs.isEmpty ? null : xs.reduce((a, b) => a + b) / xs.length;
double? minDouble(List<double> xs) =>
xs.isEmpty ? null : xs.reduce((a, b) => a < b ? a : b);
double? maxDouble(List<double> xs) =>
xs.isEmpty ? null : xs.reduce((a, b) => a > b ? a : b);

final payload = <String, dynamic>{
'day': day.date,
'sleep_samples': ts.length,
'sleep_span_sec': ts.last - ts.first,
'feature_disabled': spo2?['disabled'] == true,
'red': <String, dynamic>{
'non_zero': redNonZero,
'zero': red.length - redNonZero,
'coverage': redNonZero / red.length,
'unique': red.toSet().length,
'min': minInt(red),
'max': maxInt(red),
'mean': meanInt(red).toStringAsFixed(2),
'first10': red.take(10).toList(),
},
'ir': <String, dynamic>{
'non_zero': irNonZero,
'zero': ir.length - irNonZero,
'coverage': irNonZero / ir.length,
'unique': ir.toSet().length,
'min': minInt(ir),
'max': maxInt(ir),
'mean': meanInt(ir).toStringAsFixed(2),
'first10': ir.take(10).toList(),
},
'ratio': <String, dynamic>{
'samples': ratios.length,
'min': minDouble(ratios)?.toStringAsFixed(6),
'max': maxDouble(ratios)?.toStringAsFixed(6),
'mean': meanDouble(ratios)?.toStringAsFixed(6),
'first10': ratios.take(10).map((v) => v.toStringAsFixed(6)).toList(),
},
'odi': <String, dynamic>{
'disabled': spo2?['disabled'],
'note': spo2?['note'],
'value': spo2?['odi_per_hour'],
'dip_count': spo2?['dip_count'],
'signal_coverage': spo2?['signal_coverage'],
'trusted_coverage': spo2?['trusted_coverage'],
'confidence': spo2?['confidence'],
'reject_counts': spo2?['reject_counts'],
'severity_counts': spo2?['severity_counts'],
'debug': spo2?['debug'],
},
};
_log('[spo2-detect] ${jsonEncode(payload)}');
}

/// Skip reasons that describe a TRANSIENT failure of this particular pass
/// rather than a permanently pathological day. These must never finalize:
/// finalizing locks the day out of every future pass at this algo version.
Expand Down
29 changes: 16 additions & 13 deletions lib/compute/onehz_pipeline.dart
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
// the curve series the UI needs + indexed scalars. Survives jsonEncode.

import 'dart:math' as math;
import 'dart:typed_data';

import 'package:openstrap_analytics/onehz.dart';

Expand Down Expand Up @@ -126,8 +127,6 @@ class DayBundleInput {
final List<int> sleepHr;
final List<double> sleepRrTsMs;
final List<double> sleepRrMs;
final List<int> sleepSpo2Red;
final List<int> sleepSpo2Ir;
final List<int> sleepSkinTemp;

// ── the SINGLE-SOURCE sleep segmentation (JSON of SleepSegmentation) ──────
Expand Down Expand Up @@ -192,8 +191,6 @@ class DayBundleInput {
required this.sleepHr,
required this.sleepRrTsMs,
required this.sleepRrMs,
required this.sleepSpo2Red,
required this.sleepSpo2Ir,
required this.sleepSkinTemp,
required this.sleepJson,
required this.hypnoStages,
Expand Down Expand Up @@ -222,8 +219,6 @@ class DayBundleInput {
'sleep_hr': sleepHr,
'sleep_rr_ts_ms': sleepRrTsMs,
'sleep_rr_ms': sleepRrMs,
'sleep_spo2_red': sleepSpo2Red,
'sleep_spo2_ir': sleepSpo2Ir,
'sleep_skin_temp': sleepSkinTemp,
'sleep_json': sleepJson,
'hypno_stages': hypnoStages,
Expand All @@ -245,9 +240,20 @@ class DayBundleInput {
static DayBundleInput fromJson(Map<String, dynamic> m) {
List<int> ints(String k) =>
((m[k] as List?) ?? const []).map((e) => (e as num).toInt()).toList();
List<double> dbls(String k) => ((m[k] as List?) ?? const [])
.map((e) => (e as num).toDouble())
.toList();
// The substrate packs these as Float64List and the isolate boundary hands
// them back typed; unboxing them into a plain List<double> was re-boxing
// every element for nothing. Always a COPY, never an alias: on the direct
// (synchronous, in-test) path returning the caller's list would share the
// substrate's arrays across two repos with nobody enforcing read-only.
List<double> dbls(String k) {
final v = (m[k] as List?) ?? const [];
if (v is List<double>) return Float64List.fromList(v);
final out = Float64List(v.length);
for (var i = 0; i < v.length; i++) {
out[i] = (v[i] as num).toDouble();
}
return out;
}
List<String> strs(String k) =>
((m[k] as List?) ?? const []).map((e) => e.toString()).toList();
return DayBundleInput(
Expand All @@ -260,8 +266,6 @@ class DayBundleInput {
sleepHr: ints('sleep_hr'),
sleepRrTsMs: dbls('sleep_rr_ts_ms'),
sleepRrMs: dbls('sleep_rr_ms'),
sleepSpo2Red: ints('sleep_spo2_red'),
sleepSpo2Ir: ints('sleep_spo2_ir'),
sleepSkinTemp: ints('sleep_skin_temp'),
sleepJson: ((m['sleep_json'] as Map?) ?? const {})
.cast<String, dynamic>(),
Expand Down Expand Up @@ -456,7 +460,6 @@ Map<String, dynamic> deriveDayBundle(Map<String, dynamic> inputJson) {
const kSpo2Refusal = 'refused: the red and IR channels are one signal — '
'ir − red is a fixed offset within a session, so any ratio built from '
'them measures baseline drift, not oxygenation';
final odiTs = [for (final t in d.sleepTsSec) t.toDouble()];
const odi = Metric<RelativeOdiResult>.absent(
tier: Tier.relative,
inputs_used: ['spo2_red_raw', 'spo2_ir_raw'],
Expand Down Expand Up @@ -1044,7 +1047,7 @@ Map<String, dynamic> deriveDayBundle(Map<String, dynamic> inputJson) {
'inputs_used': const ['spo2_red_raw', 'spo2_ir_raw'],
'note': kSpo2Refusal,
'debug': <String, dynamic>{
'sleep_samples': odiTs.length,
'sleep_samples': d.sleepTsSec.length,
},
};

Expand Down
Loading
Loading