Практические руководства
Определение признаков усталости с помощью классификации последовательностей
Обучение модели классификации усталости с использованием носимых данных о состоянии здоровья
Представьте, что вы какое-то время ехали по дороге, и начало темнеть. Вы чувствуете, что ваши глаза становятся тяжелее, и знаете, что вам следует остановиться, чтобы отдохнуть, но хотите продолжить движение, чтобы вовремя добраться до места назначения ...
Я уверен, что многие люди в какой-то момент сталкивались с этой ситуацией, типичным примером сонного вождения, из-за которого у нас в 3 раза больше шансов попасть в дорожно-транспортное происшествие.
Что, если бы существовал способ определить, когда мы начинаем чувствовать усталость?
Это постановка проблемы, которую мы постараемся решить, используя мои личные данные Fitbit. Другой пример, который я написал с использованием данных Fitbit, можно найти здесь.
Задача
Мы будем использовать данные о сне и частоте пульса, полученные с моих личных часов Fitbit. Для этой задачи данные о сне дают представление о том, когда я начал спать, и могут использоваться в качестве точки для разграничения между бодрствованием и сном. Кроме того, это помогает нам оценить временной интервал для усталости; для удобства это будет определено как последние тридцать минут перед сном. Определение этих различных стадий помогает маркировать наши данные о частоте пульса, которые измеряют количество ударов сердца в минуту.
Это дает нам помеченную одномерную последовательность измерений сердечного ритма, классифицированную по интервалам сна, которая будет использоваться в качестве входных данных в нашей модели классификации последовательности. Наша модель будет ГРУ, построенным с использованием Кераса.
После создания и обучения нашего классификатора последовательностей мы увидим, насколько хорошо он работает с некоторыми невидимыми данными.
Загрузка и подготовка данных
Данные о сне и частоте пульса представляют собой смесь значений с отметками времени, хранящихся в файлах .csv или .json, которые представляют день месяца, поэтому нам необходимо загрузить все эти файлы и сложить их вместе.
#Import sleep data data and combine files
def Dataimport():
datasets = ['sleep_files *.csv']
for datatype in datasets:
file_list=[]
path = 'folder pathway'
os.chdir(path)
for file in glob.glob(datatype):
file_list.append(file)
dfs = []
for file in file_list:
data = pd.read_csv(path + file)
print('Reading: ' + str(file))
dfs.append(data)
concatenated = pd.concat(dfs, ignore_index=True)
concatenated = concatenated[['sleep_start','sleep_end']]
return concatenated
sleepdata = Dataimport()
Это дает нам таблицу с временем начала и окончания сна со значением вроде этого:
Sleep start: 2021-01-12 22:10:00 Sleep end: 2021-01-13 05:37:30
Мы можем разделить дату и время, поскольку позже нам нужно будет найти только дни, в которые записываются данные о сне и частоте сердечных сокращений:
#Splitting the date and time
def Datasplit(sleepdata):
sleepdata['date_start'] = sleepdata['sleep_start'].str.split('T', 1, expand=True)[0]
sleepdata['time_start'] = sleepdata['sleep_start'].str.split('T', 1, expand=True)[1]
sleepdata['date_end'] = sleepdata['sleep_end'].str.split('T', 1, expand=True)[0]
sleepdata['time_end'] = sleepdata['sleep_end'].str.split('T', 1, expand=True)[1]
sleepdata['start_of_sleep'] = pd.to_datetime(sleepdata['date_start'] + ' ' + sleepdata['time_start'])
sleepdata['end_of_sleep'] = pd.to_datetime(sleepdata['date_end'] + ' ' + sleepdata['time_end'])
sleepdata = sleepdata[['start_of_sleep', 'end_of_sleep']]
sleepdata = sleepdata.sort_values(by="start_of_sleep")
return sleepdata
sleepdata = Datasplit(sleepdata)
Для импорта данных частоты пульса мы можем повторно использовать ту же функцию Dataimport (), заменив имя файла на включение «heart_rate» в имя файла. Эти данные нужно будет очистить немного другим способом, так как мы хотим удалить все ненужные строки и пересчитать временные ряды с тем же разрешением, что и данные сна (то есть в минутах):
#cleaning the columns of the heart rate data
def HRclean(heart):
heart = heart.sort_values(by=”dateTime”)
heart = heart.set_index(‘dateTime’)
heart[“value”] = heart[“value”].apply(str)
heart[“value”] = heart[“value”].str.split(“{‘bpm’:”).str[1]
heart[“value”] = heart[“value”].str.split(“,”, 1, expand = True)[0]
heart[“value”] = heart[“value”].astype(int)
heart = heart.resample(‘1Min’).mean()
heart[‘value’] = heart[‘value’].round(0)
heart[‘date’] = heart.index
heart = heart[[‘date’, ‘value’]]
return heart
heartdata = Dataimport()
heart = HRclean(heartdata)
Поскольку мы определим интервал «усталости» как тридцать минут до сна, мы уже можем представить, что будет большой дисбаланс в классе по сравнению с временем бодрствования и сна. Чтобы смягчить это, мы можем отфильтровать данные частоты пульса до меньшего интервала, скажем, с раннего вечера до раннего утра:
#selecting only values in the evening (times around the tiredness since mornings are irrelevant) heart = heart.between_time(‘19:00’, ‘03:00’) heart[“only_date”] = [d.date() for d in heart[“date”]] sleepdata[“only_date”] = [d.date() for d in sleepdata[“start_of_sleep”]] #Identifying rows where sleep data exists for the given day of heart rate data heart[‘sleep_data_exists’] = pd.Series(heart.only_date.isin(sleepdata.only_date).values.astype(int), heart.date.values) heart = heart[heart[‘sleep_data_exists’] == 1]
Теперь мы можем пометить данные частоты пульса как спящие или бодрствующие в зависимости от диапазона дат и времени сна:
#for each HR row, need to see if that time was during sleep or not, and label as 1 or 0 def Labelling(heart): print(‘labelling the data…’) heart[‘sleep_label’] = 0 #for each heartrate value, for each dt range, if hr date in dt range (per row), =1 else = continue for i in range(len(heart)): print(str(i) + ‘ of ‘+ str(len(heart))) for j in range(len(sleepdata)): if heart[‘date’][i] >= sleepdata[‘start_of_sleep’][j] and heart[‘date’][i] <= sleepdata[‘end_of_sleep’][j]: heart[‘sleep_label’][i] = 1 else: continue return heart heart = Labelling(heart)
Последним этапом подготовки будет обозначение нашего «усталого» класса:
#selecting the time n rows before sleep starts
idx = heart.index.get_indexer_for(heart[heart['sleep_label'] == 1].index)
subset = heart.iloc[np.unique(np.concatenate([np.arange(max(i-30,0), min(i-30+1, len(heart)))
for i in idx]))]
subset = subset[subset.sleep_label == 0]
heart['tired_label'] = pd.Series(heart.date.isin(subset.date).values.astype(int))
#cleaning the final labels into numerical values
heart['label'] = pd.Series()
heart['label'][heart.tired_label == 1] = 2
heart['label'][heart.sleep_label == 1] = 1
heart['label'] = heart['label'].fillna(0)
heart = heart.dropna()
Изучение данных
Теперь мы можем приступить к более подробному рассмотрению нашего набора данных. Начнем с импорта полезных библиотек. Тогда неплохо было бы визуализировать некоторые помеченные нами данные о частоте пульса, чтобы мы получить представление о том, как частота пульса меняется в зависимости от класса:
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from tensorflow import keras
import matplotlib.pyplot as plt
import scipy.stats as stats
import seaborn as sns
from tensorflow.keras import backend as K
from keras.layers import Dense, Dropout
sns.set(rc={"figure.figsize":(30, 8)})
sns.scatterplot(x = df.index, y = df['heart_value'][:1500], hue = df.label, palette = 'Spectral')
plt.xlabel("Heart Rate (BPM)")
plt.ylabel("Time (Minutes)")
plt.show()
Мы ясно видим, что частота сердечных сокращений варьируется от сна к бодрствованию. Значения частоты пульса, помеченные как «усталость», являются переходными и поэтому менее отчетливыми, что сделает наше моделирование интересным. Несмотря на сокращение продолжительности собеседования, у нас все еще есть классовый дисбаланс, который необходимо решить позже.
Также неплохо посмотреть на распределение значений частоты пульса по классам:
И посмотрите их сводную статистику:
В этом случае 0 бодрствует, 1 спит и 2 устает. Класс «усталый» имеет среднее значение, подобное классу «спящий», однако имеет стандартное отклонение, аналогичное классу «бодрствования». Это перекликается с нашими наблюдениями за первым проходом, полученными при просмотре временных рядов сердечного ритма.
Предварительная обработка последовательности
Теперь давайте разделим наши данные на последовательности фиксированной длины. Наши последовательности будут длиться 30 минут, чтобы соответствовать продолжительности перерыва в работе. Нам также нужно разделить наши данные:
def create_dataset(X, y, time_steps, step): Xs, ys = [], [] for i in range(0, len(X) - time_steps, step): v = X.iloc[i:(i + time_steps)].values labels = y.iloc[i: i + time_steps] Xs.append(v) ys.append(stats.mode(labels)[0][0]) return np.array(Xs), np.array(ys).reshape(-1, 1) interval = 30 X_train_full, y_train_full = create_dataset(x1, y1, 1, interval) X_train, X_test, y_train, y_test = train_test_split(X_train_full, y_train_full, test_size=0.05, stratify = y_train_full, random_state=42, shuffle = True)
Мы можем использовать Keras для преобразования наших меток в закодированные категории и дальнейшего разделения нашего набора данных, чтобы у нас были отдельные наборы для обучения (76%), проверки (19%) и тестирования (5%)… довольно странное разделение, если оглянуться назад:
y_train1 = keras.utils.to_categorical(y_train, num_classes = None) y_test1 = keras.utils.to_categorical(y_test, num_classes = None) X_train = X_train.astype(np.float32) y_train1 = y_train1.astype(np.float32) #making the validation set separately X_train, X_val, y_train1, y_val = train_test_split(X_train, y_train1, test_size=0.2, stratify = y_train1, random_state=42, shuffle = True)
Мы используем аргумент «стратифицировать» в train_test_split (), чтобы наши метки распределялись пропорционально по классам. Чтобы помочь с дисбалансом классов, мы будем использовать веса классов, где наша модель будет весить правильное предсказание "усталости" выше, чем другие классы. Вес будет пропорционален количеству наблюдений в наборе данных:
class_weight = {3: 1., 1: 1., 2: int((sum(y_train1.iloc[:,0]) + sum(y_train1.iloc[:,2])) / sum(y_train1.iloc[:,1]))}
В этом случае наш уставший класс имеет в 13 раз больший вес, чем бодрствующий или спящий. Поскольку это мультиклассовая классификация последовательностей, мы будем использовать оценку F1 для измерения точности. Полезное руководство по этому вопросу, а также код для реализации Keras можно найти здесь.
Построение модели
Наконец, на здание модели. Как уже упоминалось, мы будем использовать режим ГРУ, поскольку он быстрее для обучения. Мы строим модель, используя Keras, как показано ниже, обязательно добавляя слои Dropout, чтобы минимизировать переоснащение, и передавая выходную последовательность от одного ГРУ к другому. Наши слои ГРУ начнутся с 32 единиц, которые будут последовательно умножаться. Мы используем оптимизатор Adam, устанавливаем активацию на softmax (поскольку у нас есть несколько классов) и измеряем перекрестную потерю энтропии:
def create_gru_model(unit): inputs = keras.Input(shape=(X_train.shape[1],X_train.shape[2])) x = layers.GRU(unit*1, activation='tanh', return_sequences=True) (inputs) x = layers.Dropout(0.25)(x) x = layers.GRU(unit*2, activation='tanh', return_sequences=True)(x) x = layers.Dropout(0.25)(x) x = layers.GRU(unit*2, activation='tanh', return_sequences=True)(x) x = layers.Dropout(0.25)(x) x = layers.GRU(unit*3, activation='tanh')(x) outputs = layers.Dense(y_train1.shape[1], activation="softmax")(x) model = keras.Model(inputs, outputs) opt = keras.optimizers.Adam(learning_rate=1e-3) model.compile(loss='categorical_crossentropy', optimizer= opt, metrics=[custom_f1]) return model model_2 = create_gru_model(32) history2 = model_2.fit(X_train, y_train1, validation_data = (X_val, y_val), epochs = 200, batch_size = 256, shuffle= False, class_weight=class_weight)
После 200 эпох обучения и проверки мы можем увидеть результат F1 и потерю модели:
sns.set(rc={"figure.figsize":(12, 12)})
plt.plot(history2.history['custom_f1'])
plt.plot(history2.history['val_custom_f1'])
plt.ylabel('F1 Score')
plt.xlabel('Epoch')
plt.legend()
plt.show()
plt.plot(history2.history['loss'])
plt.plot(history2.history['val_loss'])
plt.ylabel('Loss')
plt.xlabel('Epoch')
plt.legend()
plt.show()
Итоговая оценка проверки выше 0,95 кажется неплохой. Наконец, мы сделаем прогноз на нашем тестовом наборе и оценим результаты в матрице неточностей. Нам нужно будет использовать np.argmax (), чтобы помочь нам при работе с несколькими классами:
y_pred = np.argmax(model_2.predict(X_test), axis=1) y_pred = np.expand_dims(y_pred, axis=-1) from sklearn.metrics import plot_confusion_matrix matrix = confusion_matrix(y_test, y_pred) sns.heatmap(matrix, annot=True, fmt=’g’, yticklabels= [ ‘Awake’, ‘Asleep’, ‘Tired’], xticklabels= [ ‘Awake’, ‘Asleep’, ‘Tired’])
Похоже, что подавляющее большинство классов были предсказаны правильно. Мы видим некоторые неправильные классификации между усталостью и бодрствованием, что можно понять из средних значений и гистограмм, которые мы видели ранее.
Резюме / дальнейшие действия
Мы взяли одномерный временной ряд данных о частоте сердечных сокращений в сочетании с метками, полученными из данных о сне, чтобы создать модель классификации последовательности, способную определить, когда мы спим, бодрствуем или устали. Наша модель хорошо себя зарекомендовала, но, возможно, есть дополнительные аспекты, которые следует учесть:
- Как модель будет работать для классификации двоичных последовательностей (т.е. устала / не устала)?
- Могут ли данные, измеряющие количество сделанных шагов, помочь отличить бодрствование от усталости (при условии, что физическая активность снижается при усталости)?
- Если бы мы расширили наш временной интервал, чтобы также регистрировать усталость при пробуждении, будет ли это выглядеть так же или иначе по сравнению с «вечерней усталостью»?
Нам предстоит еще многое сделать и изучить, я буду держать вас в курсе моих достижений. Спасибо за чтение!
Примечание. Эта статья не претендует на способность прогнозировать или предотвращать несчастные случаи, связанные с усталостью, и ее следует читать и использовать только в качестве упражнения по изучению данных.
