Code
from libs.chapter4.analysis.data_loading import load_reports
from libs.chapter4.analysis.statistical_tests import compare_splitting_approaches
= load_reports() reports
As previously noted, the main output of the TUG test (i.e., completioin time) can be decomposed into several components (i.e., subphases). In the literature, there seem to be different approaches to this decomposition: some authors break down the subphases of the test into standing_up, first_walk, first_turn, second_walk, second_turn and sitting_down (Zakaria et al. 2015; Madhushri et al. 2016; Beyea et al. 2017; Coelln et al. 2019). Others consider the last two subphases into a single one combining the last turn and the sitting down activities into the turn_to_sit subphase (Salarian et al. 2010; Adame et al. 2012; Milosevic, Jovanov, and Milenković 2013; Ansai et al. 2019).
From the two splitting approaches used in the literature, we train several models with four different datasets obtained from the dataset described in Smartphone and smartwatch HAR dataset:
SEATED
, STANDING_UP
, WALKING
, TURNING
and SITTING_DOWN
activities.SEATED
, STANDING_UP
, WALKING
, TURNING
and SITTING_DOWN
activities.SEATED
, STANDING_UP
, WALKING
, TURNING
and TURN_TO_SIT
activities.SEATED
, STANDING_UP
, WALKING
, TURNING
and TURN_TO_SIT
activities.The ts datasets were generated in the previous chapter, while the tts datasets were generated using the 01_relabel.py
script:
"""Data relabelling script.
Relabels the windowed data by replacing the TURNING and SITTING_DOWN labels by the TURN_TO_SIT label. Note that
only the TURNING activities inmediately before the SITTING_DOWN activity are replaced by TURN_TO_SIT.
**Example**:
$ python 01_relabel.py --input_data_path <PATH_OF_WINDOWED_DATA> --output_data_path <PATH_TO_STORE_RELABELLED_DATA>
"""
import argparse
import numpy as np
import os
import pandas as pd
import sys
"../../..")
sys.path.append(
from alive_progress import alive_bar
from libs.common.data_loading import load_subjects_data
def relabel(gt):
= {}
relabelled_gt for subject, data in gt.items():
= np.copy(data)
data_copy = np.where(np.roll(data,1) != data)[0]
changes
for i, change in enumerate(changes):
if change == 0:
continue
if data_copy[change] == 'SITTING_DOWN':
if i+1 != len(changes):
-1]:changes[i+1]] = 'TURN_TO_SIT'
data_copy[changes[ielse:
-1]:] = 'TURN_TO_SIT'
data_copy[changes[i= data_copy
relabelled_gt[subject] return relabelled_gt
def count_data(data_collection):
= {
recount 'sp': {
'SEATED': 0,
'STANDING_UP': 0,
'WALKING': 0,
'TURNING': 0,
'TURN_TO_SIT': 0
},'sw': {
'SEATED': 0,
'STANDING_UP': 0,
'WALKING': 0,
'TURNING': 0,
'TURN_TO_SIT': 0
}
}
for source, data in data_collection.items():
for subject, subject_data in data.items():
= np.unique(subject_data, return_counts=True)
unique, counts = dict(zip(unique, counts))
value_counts 'SEATED'] += value_counts['SEATED']
recount[source]['STANDING_UP'] += value_counts['STANDING_UP']
recount[source]['WALKING'] += value_counts['WALKING']
recount[source]['TURNING'] += value_counts['TURNING']
recount[source]['TURN_TO_SIT'] += value_counts['TURN_TO_SIT']
recount[source][
= pd.DataFrame(recount).transpose()
df 'TOTAL'] = df.sum(axis=1)
df[return df.to_markdown()
def store_windowed_data(windowed_data, ground_truth, path):
def store_as_npy(path, data):
with open(path, 'wb') as f:
np.save(f, np.array(data))
with alive_bar(len(windowed_data), title=f'Storing windowed data in {path}', force_tty=True, monitor='[{percent:.0%}]') as progress_bar:
for source, subjects_data in windowed_data.items():
for subject, data in subjects_data.items():
= os.path.join(path, subject)
subject_path if not os.path.exists(subject_path):
os.makedirs(subject_path)
f'{subject}_{source}.npy'), data)
store_as_npy(os.path.join(subject_path, f'{subject}_{source}_gt.npy'), ground_truth[source][subject])
store_as_npy(os.path.join(subject_path,
progress_bar()
if __name__ == '__main__':
= argparse.ArgumentParser()
parser '--input_data_path', help='Path of input windowed data', type=str, required=True)
parser.add_argument('--output_data_path', help='Path to store the relabelled windowed data', type=str, required=True)
parser.add_argument(= parser.parse_args()
args
= load_subjects_data(args.input_data_path, 'sp', True)
sp_windowed_data, sp_gt = load_subjects_data(args.input_data_path, 'sw', True)
sw_windowed_data, sw_gt
= relabel(sp_gt)
sp_gt_relabelled = relabel(sw_gt)
sw_gt_relabelled
= {
windowed_data 'sp': sp_windowed_data,
'sw': sw_windowed_data
}
= {
gt 'sp': sp_gt_relabelled,
'sw': sw_gt_relabelled,
}
print(count_data(gt))
store_windowed_data(windowed_data, gt, args.output_data_path)
Then, for each dataset, \(100\) models were trained using \(80\%\) of the subjects as training subjects and the remaining \(20\%\) as testing subjects. This proces was executed using the 02_splitting-evaluation.py
script.
"""Splitting approach evaluation script
This script trains 100 models for each data source (smartphone, smartwatch) and splitting approach. For the training process,
a 80/20 train/test split is employed with a batch size of 20 windows during 50 epochs.
**Example**:
$ python 02_splitting-evaluation.py
--ts_data_path <PATH_OF_TURNING_SITTING_DATA>
--tts_data_path <PATH_OF_TURN_TO_SIT_DATA>
--reports_output_path <PATH_TO_STORE_REPORTS>
"""
import argparse
import sys
"../../..")
sys.path.append(
from alive_progress import alive_bar
from libs.chapter4.pipeline.training import create_trainer
from libs.common.data_loading import load_data
from libs.common.data_grouping import generate_training_and_test_sets
from libs.common.ml import generate_report
from libs.common.utils import save_json, set_seed
from sklearn.model_selection import train_test_split
= {"SEATED": 0, "STANDING_UP": 1, "WALKING": 2, "TURNING": 3, "SITTING_DOWN": 4}
TURNING_AND_SITTING_MAPPING = {"SEATED": 0, "STANDING_UP": 1, "WALKING": 2, "TURNING": 3, "TURN_TO_SIT": 4}
TURN_TO_SIT_MAPPING
= 20
BATCH_SIZE = 50
EPOCHS
def training_report_from_datasets(datasets, models_per_dataset=100):
set_seed()= create_trainer(BATCH_SIZE, EPOCHS)
trainer = {}
reports
for dataset_id, (x, y) in datasets.items():
= []
reports[dataset_id] = TURNING_AND_SITTING_MAPPING.keys() if 'turning_and_sitting' in dataset_id else TURN_TO_SIT_MAPPING.keys()
activity_names with alive_bar(models_per_dataset, title=f'Training models for dataset {dataset_id}', force_tty=True) as progress:
for i in range(models_per_dataset):
= train_test_split(list(x.keys()), test_size=0.2)
train_subjects, test_subjects
= generate_training_and_test_sets(x, y, train_subjects, test_subjects)
x_train, y_train, x_test, y_test
= trainer(x_train, y_train, verbose=0)
model = model.predict(x_test)
y_pred
reports[dataset_id].append(generate_report(y_test, y_pred, activity_names))
progress()
return reports
if __name__ == '__main__':
= argparse.ArgumentParser()
parser '--ts_data_path', help='Path of data labelled with TURNING and SITTING_DOWN activities', type=str, required=True)
parser.add_argument('--tts_data_path', help='Path of data labelled with TURN_TO_SIT activity', type=str, required=True)
parser.add_argument('--reports_output_path', help='Path to store the generated reports', type=str, required=True)
parser.add_argument(= parser.parse_args()
args
= load_data(args.ts_data_path, 'sp', True, TURNING_AND_SITTING_MAPPING)
x_sp_ts, y_sp_ts = load_data(args.ts_data_path, 'sw', True, TURNING_AND_SITTING_MAPPING)
x_sw_ts, y_sw_ts
= load_data(args.tts_data_path, 'sp', True, TURN_TO_SIT_MAPPING)
x_sp_tts, y_sp_tts = load_data(args.tts_data_path, 'sw', True, TURN_TO_SIT_MAPPING)
x_sw_tts, y_sw_tts
= {
datasets 'sw_turning_and_sitting': [x_sw_ts, y_sw_ts],
'sp_turning_and_sitting': [x_sp_ts, y_sp_ts],
'sw_turn_to_sit': [x_sw_tts, y_sw_tts],
'sp_turn_to_sit': [x_sp_tts, y_sp_tts],
}
= training_report_from_datasets(datasets)
reports 'reports.json')
save_json(reports, args.reports_output_path,
from libs.chapter4.analysis.data_loading import load_reports
from libs.chapter4.analysis.statistical_tests import compare_splitting_approaches
= load_reports() reports
Table 9.1 compares the overall accuracy and F1-scores of TURNING
and SITTING_DOWN
from one side, and TURN_TO_SIT
activities from the other side, obtained for each data source – smartwatch (sw) or smartphone (sp) – and splitting approach – turning and sitting down (ts) or turn_to_sit (tts) – from the trained models. The overall accuracy obtained with the models trained with the ts is statistically better than the ones trained with the tts datasets. Moreover, the F1-score of the TURNING
activity is statistically worse in the tts datasets due to the reduced number of training samples for that activity compared with the ts datasets, caused by the fact that the TURN_TO_SIT
activity includes the TURNING
activity (which is one of the other activities to be individually detected). In addition, the F1-score of the TURN_TO_SIT
activity is low compared with the scores of TURNING
and SITTING_DOWN
in the ts datasets.
def get_accuracy_and_f1_scores(reports):
= {}
results for dataset_key, dataset_reports in reports.items():
= dataset_key.split('_')[0]
source if source not in results:
= {}
results[source]
= []
dataset_accuracies = []
dataset_turning = []
dataset_specific
= ('SITTING_DOWN', 'f1-sitting-down', 'ts') if 'turning_and_sitting' in dataset_key else ('TURN_TO_SIT', 'f1-turn_to_sit', 'tts')
specific_act, specific_score, key
for dataset_report in dataset_reports:
'accuracy'])
dataset_accuracies.append(dataset_report['TURNING']['f1-score'])
dataset_turning.append(dataset_report['f1-score'])
dataset_specific.append(dataset_report[specific_act][
= {
results[source][key] 'accuracy': dataset_accuracies,
'f1-turning': dataset_turning,
f'{specific_score}': dataset_specific
}
return results
= get_accuracy_and_f1_scores(reports)
results = compare_splitting_approaches(results, ['accuracy', 'f1-turning', 'f1-sitting-down', 'f1-turn_to_sit'])
comparison comparison
TURNING
, SITTING_DOWN
and TURN_TO_SIT
for each data source and splitting approach.
source | metric | turning_sitting | turn_to_sit | two-tailed test | |
---|---|---|---|---|---|
0 | sw | accuracy | 0.848 | 0.809 | t(198)=13.459315696940894, p-val=0.0, power=1.0 |
1 | sw | f1-turning | 0.795 | 0.565 | t(198)=44.34041443328367, p-val=0.0, power=1.0 |
2 | sw | f1-sitting-down | 0.804 | - | - |
3 | sw | f1-turn_to_sit | - | 0.735 | - |
4 | sp | accuracy | 0.857 | 0.789 | U=9036.0, p-val=0.0, power=1 |
5 | sp | f1-turning | 0.846 | 0.529 | t(198)=57.06501597390656, p-val=0.0, power=1.0 |
6 | sp | f1-sitting-down | 0.753 | - | - |
7 | sp | f1-turn_to_sit | - | 0.655 | - |
From these results, we conclude that more accurate results are obtained when considering TURNING
and SITTING_DOWN
as separate activities, compared to combining them, since not only the overall accuracy of the prediction model is better, but also the predictability for the TURNING
activity. Therefore, the first approach (separate activities) will be used in the implementation and evaluation of the system.
The documentation of the Python functions employed in this section can be found in Chapter 4 reference: