智能穿戴设备如何通过AI实现慢性病管理：从理论到完整代码落地

# train_afib.py
import torch, torchaudio, pytorch_lightning as pl
from torch import nn
from torch.utils.data import DataLoader
from sklearn.metrics import roc_auc_score

class AFibModel(pl.LightningModule):
    def __init__(self, in_ch=1, hidden=64):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Conv1d(in_ch, 32, 7, stride=2, padding=3),
            nn.ReLU(),
            nn.Conv1d(32, hidden, 5, stride=2, padding=2),
            nn.ReLU()
        )
        self.lstm = nn.LSTM(hidden, hidden, 2, batch_first=True, dropout=0.3)
        self.head = nn.Linear(hidden, 1)

    def forward(self, x):  # x: (B, 1, 512)
        z = self.encoder(x).transpose(1, 2)  # (B, T, H)
        z, _ = self.lstm(z)
        return torch.sigmoid(self.head(z[:, -1, :]))

    def training_step(self, batch, _):
        x, y = batch
        y_hat = self(x).squeeze()
        loss = nn.BCELoss()(y_hat, y.float())
        self.log("train_loss", loss)
        return loss

    def configure_optimizers(self):
        return torch.optim.Adam(self.parameters(), 1e-3)

if __name__ == "__main__":
    train_ds = torchaudio.datasets.IAFIB(root="data", split="train", download=True)
    val_ds   = torchaudio.datasets.IAFIB(root="data", split="val")
    trainer = pl.Trainer(max_epochs=20, gpus=1)
    trainer.fit(AFibModel(), DataLoader(train_ds, 256, True),
                DataLoader(val_ds, 512))
    torch.save(AFibModel().state_dict(), "afib.pt")

import torch.quantization as Q
model = AFibModel()
model.load_state_dict(torch.load("afib.pt"))
model.eval()
qmodel = Q.quantize_dynamic(model, {nn.LSTM, nn.Linear}, dtype=torch.qint8)
torch.jit.save(torch.jit.script(qmodel), "afib_q8.pt")
!python -m tf.lite.python.tflite_convert \
        --saved_model_dir=afib_q8.pt --output_file=afib.tflite

// main.cpp (Arduino / Zephyr)
#include <TensorFlowLite.h>
#include "afib_model_data.h"  // 由 xxd -i afib.tflite > afib_model_data.h 生成

tflite::MicroErrorReporter micro_reporter;
tflite::MicroInterpreter interpreter(
    tflite::GetModel(afib_model_data), 
    resolver, tensor_arena, kTensorArenaSize, &micro_reporter);

void loop() {
  float ppg[512];
  read_ppg_dma(ppg);  // 传感器 4 ms 采样
  input->data.f = ppg;
  interpreter.Invoke();
  float prob = output->data.f[0];
  if (prob > 0.7f) send_ble_alert("AFIB");
}

# client.py
import flwr as fl
class AFibClient(fl.client.NumPyClient):
    def get_parameters(self): return [val.cpu().numpy() for val in model.parameters()]
    def fit(self, parameters, config):
        set_parameters(model, parameters)
        trainer.fit(model, local_loader)
        return [val.cpu().numpy() for val in model.parameters()], len(local_loader), {}
    def evaluate(self, parameters, config):
        set_parameters(model, parameters)
        loss, y_hat, y = trainer.validate()
        return float(loss), len(val_loader), {"auc": roc_auc_score(y, y_hat)}
fl.client.start_numpy_client("0.0.0.0:8080", AFibClient())