跳转到主要内容

概览

数据映射(data mapping)把 vendor 特定的 JSON 响应转换为 Mirobody Health 的标准格式,从而确保所有健康数据源的一致性。

标准格式

All health data is transformed into StandardPulseData objects:
from mirobody.pulse.theta.platform.indicator import StandardPulseData, StandardIndicator

data = StandardPulseData(
    indicator=StandardIndicator.HEART_RATE,
    value="72",
    start_time="2024-01-15T10:00:00Z",
    end_time="2024-01-15T10:00:00Z",
    source="garmin"
)

映射模式

Basic Mapping

将简单值直接映射:
INDICATOR_MAPPING = {
    "daily_steps": StandardIndicator.DAILY_STEPS,
    "sleep_seconds": StandardIndicator.DAILY_SLEEP_DURATION,
    "resting_heart_rate": StandardIndicator.DAILY_HEART_RATE_RESTING,
}

def format_data(self, raw_data: Dict) -> List[StandardPulseData]:
    """Transform vendor data"""
    results = []
    
    for vendor_key, standard_indicator in INDICATOR_MAPPING.items():
        if vendor_key in raw_data:
            results.append(StandardPulseData(
                indicator=standard_indicator,
                value=str(raw_data[vendor_key]),
                start_time=raw_data["date"],
                end_time=raw_data["date"]
            ))
    
    return results

Nested Data Mapping

针对嵌套 JSON 结构:
def format_sleep_data(self, sleep_response: Dict) -> List[StandardPulseData]:
    """Map nested sleep data"""
    results = []
    sleep_data = sleep_response.get("sleep", {})
    
    # Total sleep duration
    if "total_seconds" in sleep_data:
        results.append(StandardPulseData(
            indicator=StandardIndicator.DAILY_SLEEP_DURATION,
            value=str(sleep_data["total_seconds"] * 1000),  # Convert to ms
            start_time=sleep_data["start_time"],
            end_time=sleep_data["end_time"]
        ))
    
    # Sleep stages
    stages = sleep_data.get("stages", {})
    if "deep_seconds" in stages:
        results.append(StandardPulseData(
            indicator=StandardIndicator.DAILY_DEEP_SLEEP,
            value=str(stages["deep_seconds"] * 1000),
            start_time=sleep_data["start_time"],
            end_time=sleep_data["end_time"]
        ))
    
    return results

Unit Conversion

将 vendor 单位转换为标准单位:
# Standard units: grams, milliseconds, meters
UNIT_CONVERSIONS = {
    "kg_to_grams": lambda x: x * 1000,
    "seconds_to_ms": lambda x: x * 1000,
    "minutes_to_ms": lambda x: x * 60 * 1000,
}

# Example
weight_kg = raw_data["weight"]
weight_grams = UNIT_CONVERSIONS["kg_to_grams"](weight_kg)

Standard Indicators

Activity Indicators

Standard IndicatorUnitDescription
DAILY_STEPSsteps每日总步数
DAILY_DISTANCEmeters行走距离
DAILY_CALORIES_ACTIVEkcal活动消耗热量
DAILY_FLOORS_CLIMBEDfloors爬楼层数
ACTIVE_TIMEmilliseconds活动时长

Sleep Indicators

Standard IndicatorUnitDescription
DAILY_SLEEP_DURATIONmilliseconds总睡眠时长
DAILY_DEEP_SLEEPmilliseconds深睡时长
DAILY_LIGHT_SLEEPmilliseconds浅睡时长
DAILY_REM_SLEEPmillisecondsREM 睡眠时长
SLEEP_EFFICIENCYpercentage睡眠效率

Heart Rate Indicators

Standard IndicatorUnitDescription
HEART_RATEbpm瞬时心率
DAILY_HEART_RATE_RESTINGbpm静息心率
DAILY_AVG_HEART_RATEbpm当日平均心率
HRVms心率变异性

完整参考

如需完整标准指标列表与更详细的映射示例:

Full Provider Integration Guide

查看完整 data mapping 模式、单位转换与指标参考

下一步