Wykres EKG i oddechu w Pythonie

Narysuj EKG i oddech na wspólnej osi czasu z Aidlab Python SDK i matplotlib. Zapisz próbki w CSV i pobierz pomiary z pamięci urządzenia.

W tym poradniku narysujesz na żywo EKG i krzywą oddechową z Aidlaba. Zapiszesz też próbki w pliku CSV. Znaczniki czasu z urządzenia pozwolą umieścić oba sygnały na wspólnej osi czasu.

Użyjemy Aidlab Python SDK i matplotlib. Na końcu pokażemy również, jak pobrać pomiary zapisane w pamięci urządzenia.

Czego potrzebujesz

Potrzebujesz Pythona 3.10 lub nowszego, komputera z Bluetooth 4.0+ i Aidlaba obsługującego EKG oraz pomiar oddechu. Do zapisu offline potrzebny jest model z pamięcią pomiarów. Sprawdź pojemności pamięci Aidlab 1.

Uruchom skrypt na komputerze, na którym matplotlib może otworzyć okno wykresu, na przykład przez TkAgg, QtAgg lub MacOSX. Zainstaluj potrzebne pakiety:

python -m pip install aidlabsdk matplotlib

Załóż pas tak, aby elektrody stykały się ze skórą. Próbki pojawią się, gdy urządzenie wykryje, że jest założone. Jeśli w pobliżu jest kilka urządzeń, wybierz swoje z wyników skanowania zamiast polegać na pierwszej pozycji.

Co zawierają oba sygnały

Sygnał	Wartość z SDK	Co pokazuje wykres
DataType.ECG	Przebieg elektryczny w woltach	Te same próbki przeliczone na miliwolty do wyświetlenia
DataType.RESPIRATION	Przebieg impedancji w omach	Zmiany krzywej oddechowej w czasie

Krzywa oddechowa pokazuje zmiany impedancji klatki piersiowej podczas oddychania. Aby uzyskać liczbę oddechów na minutę, osobno wybierz DataType.RESPIRATION_RATE. Amplituda krzywej jest pomiarem impedancji, a nie skalibrowaną objętością powietrza w litrach. Zobacz typy danych i częstotliwości próbkowania.

Kompletny wykres na żywo z eksportem CSV

Zapisz poniższy kod jako respiration_plotting.py. Zbiera oba sygnały, rysuje ostatnie dziesięć sekund każdego z nich i zapisuje wszystkie odebrane próbki w nowym pliku CSV.

import asyncio
import csv
from collections import deque
from datetime import datetime

import matplotlib.pyplot as plt
from aidlab import AidlabManager, DataType, DeviceDelegate


class Recorder(DeviceDelegate):
    def __init__(self, writer):
        self.writer = writer
        self.samples = {"ecg": deque(), "respiration": deque()}
        self.first_timestamp = None

    def add_sample(self, signal, timestamp, value):
        self.writer.writerow([signal, timestamp, value, "live"])
        if self.first_timestamp is None:
            self.first_timestamp = timestamp
        samples = self.samples[signal]
        samples.append((timestamp, value))
        while samples and timestamp - samples[0][0] > 10_000:
            samples.popleft()

    def did_receive_ecg(self, device, timestamp, value):
        self.add_sample("ecg", timestamp, value)

    def did_receive_respiration(self, device, timestamp, value):
        self.add_sample("respiration", timestamp, value)

    def did_disconnect(self, device, reason):
        print("Disconnected:", reason)


async def main():
    devices = await AidlabManager().scan()
    if not devices:
        print("No device found. Turn Aidlab on and check Bluetooth.")
        return

    device = devices[0]
    print("Connecting to:", device.name, device.address)
    filename = f"signals-{datetime.now():%Y%m%d-%H%M%S-%f}.csv"
    plt.ion()
    figure, axes = plt.subplots(2, 1, sharex=True, layout="constrained")
    lines = [axis.plot([], [])[0] for axis in axes]
    axes[0].set_ylabel("ECG (mV)")
    axes[1].set_ylabel("Respiration (Ω)")
    axes[1].set_xlabel("Time since first sample (s)")
    plt.show(block=False)

    with open(filename, "x", newline="") as output:
        writer = csv.writer(output)
        writer.writerow(["signal", "timestamp_ms", "value", "source"])
        recorder = Recorder(writer)
        try:
            await device.connect(recorder)
            signals = [DataType.ECG, DataType.RESPIRATION]
            await device.collect(signals, [])
            print("Recording to:", filename, "— close the plot to finish.")
            while device.is_connected and plt.fignum_exists(figure.number):
                for axis, line, signal, scale in zip(
                    axes, lines, ("ecg", "respiration"), (1000, 1)
                ):
                    samples = recorder.samples[signal]
                    if samples:
                        line.set_data(
                            [(ts - recorder.first_timestamp) / 1000 for ts, _ in samples],
                            [value * scale for _, value in samples],
                        )
                        axis.relim()
                        axis.autoscale_view()
                output.flush()
                figure.canvas.draw_idle()
                figure.canvas.flush_events()
                await asyncio.sleep(0.1)
        finally:
            await device.disconnect()
            plt.close(figure)

    print("Saved:", filename)


if __name__ == "__main__":
    asyncio.run(main())

Uruchom go z terminala:

python respiration_plotting.py

Górny wykres pokazuje EKG, a dolny — oddech. Krzywa oddechowa będzie podobna do dotychczasowego przykładu poniżej; jej kształt i amplituda zależą od kontaktu pasa ze skórą, oddychania i ruchu.

Wykres odświeża się około dziesięciu razy na sekundę. Ten odstęp steruje wyświetlaniem, a nie częstotliwością próbkowania urządzenia. Aktualizujemy dwa istniejące obiekty linii; kolejki przechowują tylko dziesięciosekundowe okno, a CSV — całą sesję. Inne sposoby rysowania opisuje przewodnik po interaktywnych wykresach matplotlib.

Wspólna oś czasu dla EKG i oddechu

Do umieszczenia próbki na osi czasu używaj jej timestamp, wyrażonego w milisekundach czasu Unix. Czas wywołania callbacka może różnić się od czasu pomiaru, ponieważ próbki mogą być dostarczane partiami.

Traktuj oba sygnały jako osobne serie. Ich częstotliwości próbkowania mogą się różnić, więc łączenie próbek EKG i oddechu według indeksu na liście zniekształciłoby oś czasu. W tym przykładzie oba wykresy mają wspólny początek czasu i oś x. Nie interpolujemy ani nie zmieniamy częstotliwości próbkowania.

W pliku CSV próbki EKG pozostają w woltach, a sygnał oddechowy — w omach. Plik zawiera następujące kolumny:

Kolumna	Znaczenie
signal	ecg lub respiration
timestamp_ms	Czas pomiaru w milisekundach czasu Unix
value	Oryginalna wartość z SDK
source	live albo memory po dodaniu rozszerzenia poniżej

Przed analizą podziel próbki na EKG i oddech, a każdą serię posortuj według timestamp_ms. Wiersze są zapisywane w kolejności wywołań callbacków, więc cały plik nie musi być uporządkowany chronologicznie.

Zapis EKG i oddechu bez komputera

Dwie listy przekazywane do collect pełnią różne funkcje: pierwsza wybiera sygnały przesyłane na żywo, a druga — zapisywane w pamięci urządzenia po rozłączeniu. Aby ustawić zapis offline EKG i oddechu, zastąp linię rozpoczynającą pomiar:

await device.collect(signals, signals)

Po wykonaniu polecenia zamknij wykres, aby się rozłączyć, i pozostaw Aidlab założony podczas zapisu. Ustawienia zapisu offline pozostają w urządzeniu po rozłączeniu.

Czas zapisu zależy od pamięci urządzenia i wybranych sygnałów. Około ośmiu godzin dla wariantu Aidlab 1 z pamięcią 64 MB dotyczy konfiguracji oficjalnej aplikacji mobilnej; własny zapis EKG i oddechu ma inne zapotrzebowanie na pamięć. Zobacz pojemność pamięci wewnętrznej.

Pobieranie zapisanego pomiaru

Przed ponownym połączeniem dodaj te metody wewnątrz klasy Recorder:

    def did_receive_past_ecg(self, device, timestamp, value):
        self.writer.writerow(["ecg", timestamp, value, "memory"])

    def did_receive_past_respiration(self, device, timestamp, value):
        self.writer.writerow(["respiration", timestamp, value, "memory"])

    def sync_state_did_change(self, device, state):
        print("Synchronization:", state.name)

Przy kolejnym uruchomieniu wróć do pomiaru na żywo przez await device.collect(signals, []). Bezpośrednio po tej linii dodaj:

await device.start_synchronization()

Wykres na żywo nadal działa, a zapisane wcześniej próbki trafiają do CSV z source=memory. Pozostaw okno otwarte do komunikatu synchronizacji END; EMPTY oznacza brak zapisanego pomiaru do pobrania. Stan STOP lub UNAVAILABLE nie oznacza zakończenia pobierania.

W firmware 3.7.84–3.7.110 rozpoczęcie pomiaru automatycznie uruchamia synchronizację: pomiń osobne wywołanie start_synchronization(). Przewodnik synchronizacji opisuje ten wyjątek i stany transferu. Pobranie danych i czyszczenie pamięci urządzenia to osobne operacje; ten przykład nie usuwa zapisanych próbek.

Co możesz zbudować dalej

Jeśli tworzysz aplikację sterowaną oddechem, zobacz Breathing Sky i poradnik pomiaru oddechu w Unity. Do analizy odstępów między uderzeniami wybierz pochodzące z EKG DataType.RR i przeczytaj jak mierzyć HRV.

Jeśli wybierasz sprzęt do takiego projektu, porównaj Aidlab i Polar lub Aidlab i Shimmer, a następnie sprawdź sygnały i pamięć konkretnego urządzenia.