血压分析

import time
import appui
import health

READ_TYPES = ["blood_pressure_systolic", "blood_pressure_diastolic"]


def sample_records():
    now = time.time()
    rows = []
    values = [(118, 76), (121, 78), (116, 74), (124, 80), (119, 77), (122, 79), (117, 75)]
    for index, pair in enumerate(values):
        ts = now - (len(values) - index) * 86400
        rows.append({
            "systolic": pair[0],
            "diastolic": pair[1],
            "unit": "mmHg",
            "start": ts,
            "end": ts,
        })
    return rows


initial_records = sample_records()

state = appui.State(
    days="30",
    records=initial_records,
    summary=health.summarize_blood_pressure(initial_records),
    source="示例数据",
    loading=False,
    message="真机授权后可读取 HealthKit 血压数据",
)


def date_label(timestamp):
    return time.strftime("%m/%d %H:%M", time.localtime(float(timestamp)))


def load_health_data():
    state.loading = True
    end = time.time()
    start = end - int(state.days) * 86400
    records = []

    if health.is_available():
        health.request_authorization(read=READ_TYPES)
        records = health.query_blood_pressure(start=start, end=end)

    if records:
        source = "HealthKit"
        message = f"读取 {len(records)} 条血压记录"
    else:
        records = sample_records()
        source = "示例数据"
        message = "未读取到 HealthKit 血压记录，当前显示示例数据"

    state.batch_update(
        records=records,
        summary=health.summarize_blood_pressure(records),
        source=source,
        message=message,
        loading=False,
    )


def set_days(value):
    state.days = value
    load_health_data()


def record_start(item):
    return float(item.get("start", 0))


def chart_rows(field):
    rows = sorted(state.records, key=record_start)[-14:]
    return [
        {
            "day": time.strftime("%m/%d", time.localtime(float(item.get("start", 0)))),
            "value": float(item.get(field, 0)),
        }
        for item in rows
    ]


def stat_text(field, label):
    summary = state.summary or {}
    stats = summary.get(field, {})
    if not stats:
        return appui.LabeledContent(label, value="--")
    return appui.LabeledContent(
        label,
        value=f'{stats["avg"]:.0f} / {stats["min"]:.0f}-{stats["max"]:.0f} mmHg',
    )


def record_row(item):
    return appui.HStack(
        [
            appui.VStack(
                [
                    appui.Text(date_label(item.get("start", 0))).font("headline"),
                    (
                        appui.Text(item.get("unit", "mmHg"))
                        .font("caption")
                        .foreground_color("secondaryLabel")
                    ),
                ],
                alignment="leading",
            ),
            appui.Spacer(),
            (
                appui.Text(
                    f'{item.get("systolic", 0):.0f}/'
                    f'{item.get("diastolic", 0):.0f}'
                )
                .font("title3")
                .bold()
            ),
        ],
        spacing=12,
    )


def record_key(item):
    return item.get("start", 0)


def body():
    latest = (state.summary or {}).get("latest") or {}
    return appui.NavigationStack(
        appui.List(
            [
                appui.Section(
                    [
                        appui.LabeledContent("数据源", value=state.source),
                        appui.LabeledContent("状态", value=state.message),
                        appui.Picker(
                            "时间范围",
                            selection=state.days,
                            options=["7", "30", "90"],
                            on_change=set_days,
                        ).picker_style("segmented"),
                        appui.Button("刷新 HealthKit", action=load_health_data)
                            .button_style("bordered_prominent"),
                    ],
                    header="同步",
                    footer="本示例只做个人数据统计和可视化，不提供医疗诊断。",
                ),
                appui.Section(
                    [
                        appui.LabeledContent(
                            "最近一次",
                            value=(
                                f'{latest.get("systolic", 0):.0f}/'
                                f'{latest.get("diastolic", 0):.0f} mmHg'
                            )
                            if latest else "--",
                        ),
                        stat_text("systolic", "收缩压 均值/范围"),
                        stat_text("diastolic", "舒张压 均值/范围"),
                    ],
                    header="摘要",
                ),
                appui.Section(
                    [
                        appui.Chart(
                            chart_rows("systolic"),
                            x="day",
                            y="value",
                            type="line",
                            color="systemRed",
                        )
                            .frame(height=180),
                        appui.Chart(
                            chart_rows("diastolic"),
                            x="day",
                            y="value",
                            type="line",
                            color="systemBlue",
                        )
                            .frame(height=180),
                    ],
                    header="趋势",
                ),
                appui.Section(
                    [appui.ForEach(state.records[:20], record_row, key=record_key)],
                    header="记录",
                ),
            ]
        )
        .refreshable(load_health_data)
        .navigation_title("血压")
    )


appui.run(body, state=state, presentation="sheet")